ldap/admin/src/scripts/repl-monitor.pl.in | 271 ++++++--- ldap/servers/plugins/replication/repl5.h | 13 ldap/servers/plugins/replication/repl5_agmt.c | 472 +++++++++++++++- ldap/servers/plugins/replication/repl5_agmtlist.c | 8 ldap/servers/plugins/replication/repl5_plugins.c | 7 ldap/servers/plugins/replication/repl5_protocol.c | 2 ldap/servers/plugins/replication/repl5_replica.c | 43 + ldap/servers/plugins/replication/repl5_replica_config.c | 58 - ldap/servers/plugins/replication/repl5_ruv.c | 1 ldap/servers/plugins/replication/repl_globals.c | 1 ldap/servers/slapd/entry.c | 6 ldap/servers/slapd/rdn.c | 10 ldap/servers/slapd/slapi-plugin.h | 10 13 files changed, 752 insertions(+), 150 deletions(-)
New commits: commit c30c897acb80ae00ecdd3063145ddd716caa96f1 Author: Mark Reynolds mreynolds@redhat.com Date: Thu Oct 31 12:36:14 2013 -0400
Ticket 47368 - IPA server dirsrv RUV entry data excluded from replication
Bug Description: If using fractional replication agreements and stripping attributes, it's possible that no modificatins could be replicated, but the local RUV gets updated. This makes it impossible to detect if a consumer is up to date.
Fix Description: Added new attribute to db tombstone entry: nsds5AgmtMaxCSN.
nsds5AgmtMaxCSN: <repl area>:<agmt name>:<host>:<port>:<consumer rid>:<maxcsn> nsds5AgmtMaxCSN: dc=example,dc=com;to replica 1;localhost.localdomain;389;111;5270095c0000014d0000
Now, each agmt stores the maxcsn (that it should replicate, as different agmts might be stripping different attributes) in the local db tombstone. This maxcsn can be used to check the replicas tombstone RUV to check if it has seen all the expected updates.
repl-monitor.pl has been updated to process the the new agmt maxcsn in its report. The script also incorrectly reported on agreements from other replicas - this has been fixed. Also re-worded the report terminology: changed "Receiver" to "Consumer", and "Master" to "Supplier". Also improved the text report output as well. Examples of the new output(text and html) can be found in the ticket(see attachments). The script is backwards compatible with older versions of Directory Server.
https://fedorahosted.org/389/ticket/47368
Reviewed by: lkrispen & rmeggins (Thanks!!)
diff --git a/ldap/admin/src/scripts/repl-monitor.pl.in b/ldap/admin/src/scripts/repl-monitor.pl.in index 45d58a1..0ded096 100755 --- a/ldap/admin/src/scripts/repl-monitor.pl.in +++ b/ldap/admin/src/scripts/repl-monitor.pl.in @@ -153,6 +153,9 @@ # # -W Prompt for connection passwords. # +# -?, --help +# Print usage. +# # -v Print out the version of this script # # DIAGNOSTICS: @@ -185,7 +188,7 @@ my $usage = "\nusage: $0 [-f configuration-file | --configfile configuration-fil "[-c connection, --conn connection] [-a alias, --alias alias] [-k color, --color color] " . "[-h host, --host host] [-p port, --port port] [-r, --skip-header] [-s, --text] " . "[-u refresh-url, --url refresh-url] [-t refresh-interval, --interval refresh-interval ] " . - "[-W, --prompt]\n\nor : $0 -v | --version\n"; + "[-W, --prompt] [-?, --help] [-v | --version]\n";
use Getopt::Long; # parse command line arguments use Mozilla::LDAP::Conn; # LDAP module for Perl @@ -198,38 +201,43 @@ use Time::Local; # to convert GMT Z strings to localtime # my $product = "Directory Server Replication Monitor"; my $version = "Version 1.1"; -# + # ldap servers given or discovered from the replication agreements: my @servers; # = (host:port=shadowport:binddn:password:cert_db) my $serveridx; -# + # entries read from the connection section of the configuration file: my @allconnections; # = (host:port=shadowport:binddn:password:cert_db) -# + # aliases of ldap servers read from the configuration file: my %allaliases; # = {$host:$port} = (alias) + # colors my %allcolors; my @colorkeys;
-# # replicas discovered on all ldap servers my @allreplicas; # = (server#:replicaroot:replicatype:serverid:replicadn) -# + # ruvs retrieved from all replicas my %allruvs; # = {replica#:masterid} = (rawcsn:decimalcsn;mon/day/year hh:mi:ss) -# + # agreements discovered on all ldap supplier servers: my @allagreements; # = (supplier_replica#:consumer#:conntype:schedule:status) # the array may take another format after the consumer replicas are located: # @allagreements; # = (supplier_replica#:consumer_replica#:conntype:schedule:status) -# -my %ld; # ldap connection hash -# + +# agmt maxcsns hash +my %agmtmaxcsn = (); + +# ldap connection hash +my %ld; + my ($opt_f, $opt_h, $opt_p, $opt_u, $opt_t, $opt_r, $opt_s); my (@conns, @alias, @color); my ($section, $interval, $nowraw, $now, $mm, $dd, $tt, $yy, $wday); my ($fn, $rc, $prompt, $last_sidx); +my $supplierUrl = ""; my %passwords = (); my $passwd = ""; $prompt = ""; @@ -252,7 +260,8 @@ $prompt = ""; 'W|prompt' => sub { $prompt = "yes"; }, 'r|skip-header' => sub { $opt_r = "1"; }, 's|text' => sub {$opt_s = "1"; }, - 'v|version' => sub { print "$product - $version\n"; exit ;} + 'help|?' => sub { print $usage; exit 0;}, + 'v|version' => sub { print "$product - $version\n"; exit 0;} ) or die "Usage error: $usage\n";
exit -1 if &validateArgs < 0; @@ -281,6 +290,7 @@ $prompt = "";
# Start with the given host and port # The index names in %ld are defined in Mozilla::LDAP::Utils::ldapArgs() + &set_server_params(); &add_server ("$ld{host}:$ld{port}:$ld{bind}:$ld{pswd}:$ld{cert}");
$serveridx = 0; @@ -332,6 +342,7 @@ sub read_cfg_file # process the command line config params @allconnections = @conns; + if($#alias >= 0){ foreach $tmp (@alias){ $tmp =~ m/^\s*(\S.*)\s*=\s*(\S+)/; @@ -442,14 +453,15 @@ sub get_replicas # Get ruv for each replica # for ($ridx = $myridx; $ridx <= $#allreplicas; $ridx++) { - + my @agmtParts; + $replicaroot = $1 if ($allreplicas[$ridx] =~ /^\d+:([^:]*)/); # do a one level search with nsuniqueid in the filter - this will force the use of the # nsuniqueid index instead of the entry dn index, which seems to be unreliable in # heavily loaded servers $ruv = $conn->search($replicaroot, "one", "(&(nsuniqueid=ffffffff-ffffffff-ffffffff-ffffffff)(objectClass=nsTombstone))", - 0, qw(nsds50ruv nsruvReplicaLastModified)); + 0, qw(nsds50ruv nsruvReplicaLastModified nsds5AgmtMaxCSN)); next if !$ruv; # this should be an error case . . .
for ($ruv->getValues('nsds50ruv')) { @@ -459,6 +471,18 @@ sub get_replicas $allruvs {"$ridx:$masterid"} = "$2:$maxcsn"; } } + + for ($ruv->getValues('nsds5AgmtMaxCSN')) { + # nsds5AgmtMaxCSN = "replica index(ridx);suffix;agmtname;host;port;rid;maxcsn" + @agmtParts = split ( ";", $_); + $agmtParts[0] =~ s/ //; # remove spaces + $agmtParts[0] =~ lc $agmtParts[0]; + if($agmtParts[4] eq "unavailable"){ + $agmtmaxcsn{"$ridx;$agmtParts[0];$agmtParts[1];$agmtParts[2];$agmtParts[3]"} = "Unavailable"; + } else { + $agmtmaxcsn{"$ridx;$agmtParts[0];$agmtParts[1];$agmtParts[2];$agmtParts[3]"} = $agmtParts[5]; + } + }
for ($ruv->getValues('nsruvReplicaLastModified')) { if (m/{replica\s+(\d+).+?}\s*(\S+)/i) { @@ -491,7 +515,7 @@ sub get_replicas nsds5ReplicaLastInitEnd nsds5replicaReapActive nsds5replicaLastUpdateStart nsds5replicaLastUpdateEnd nsds5replicaChangesSentSinceStartup nsds5replicaLastUpdateStatus - nsds5ReplicaHost + nsds5ReplicaHost nsDS5ReplicaRoot nsds5ReplicaPort nsDS5ReplicaBindMethod nsds5ReplicaUpdateSchedule); $agreement = $conn->search("$replicadn", "sub", "(objectClass=nsDS5ReplicationAgreement)", 0, @attrlist); @@ -502,7 +526,6 @@ sub get_replicas $host = ($agreement->getValues('nsDS5ReplicaHost'))[0]; $port = ($agreement->getValues('nsDS5ReplicaPort'))[0]; $cidx = &add_server ("$host:$port"); - for (@attrlist) { $agmt{$_} = ($agreement->getValues($_))[0]; } @@ -512,10 +535,10 @@ sub get_replicas if (!$agmt{nsds5ReplicaUpdateSchedule} || ($agmt{nsds5ReplicaUpdateSchedule} eq '0000-2359 0123456') || ($agmt{nsds5ReplicaUpdateSchedule} eq '*') || - ($agmt{nsds5ReplicaUpdateSchedule} eq '* *')) { - $agmt{nsds5ReplicaUpdateSchedule} = 'always in sync'; + ($agmt{nsds5ReplicaUpdateSchedule} eq '* *')) + { + $agmt{nsds5ReplicaUpdateSchedule} = 'always in sync'; } - $agmt{ridx} = $ridx; $agmt{cidx} = $cidx; push @allagreements, %agmt; @@ -565,7 +588,7 @@ sub find_consumer_replicas for ($j = 0; $j <= $#allreplicas; $j++) {
# Get a replica on consumer - # I'm not sure what's going on here, but possibly could be made + # I'm not sure what's going on here, but possibly could be made # much simpler with normalizeDN and/or ldap_explode_dn if ($allreplicas[$j] =~ /^$c_sidx:([^:]*)/) { $val = $1; @@ -600,7 +623,6 @@ sub process_suppliers # Handle masters and hubs if ($allreplicas[$ridx] =~ /:master:(\d+):/i) { $mid = $1; - # Skip replicas without agreements defined yet next if (! grep {$_->{ridx} == $ridx} @allagreements); $maxcsn = &print_master_header ($ridx, $mid); @@ -615,7 +637,7 @@ sub process_suppliers # Skip replicas without agreements defined yet next if (! grep {$_->{ridx} == $ridx} @allagreements);
- foreach my $key (keys %allruvs) { + foreach my $key (keys %allruvs) { if ( $key =~ /$ridx:/) { my ($myridx, $mymid) = split ( /:/, "$key" ); $maxcsn = &print_hub_header($myridx, $mymid); @@ -634,6 +656,9 @@ sub process_suppliers print "<p>The server is not a master or a hub or it has no replication agreement\n"; } } + if($opt_s){ + print "\n"; + } }
sub print_master_header @@ -643,6 +668,7 @@ sub print_master_header my ($maxcsnval) = split ( /;/, "$myruv" ); my ($maxcsn) = &to_string_csn ($maxcsnval); my ($sidx, $replicaroot, $replicatype, $serverid) = split (/:/, $allreplicas[$ridx]); + my ($sline, $slen, $ii);
if ( $maxcsn eq "" ) { return $maxcsn; @@ -650,22 +676,30 @@ sub print_master_header
# Print the master name if ( $last_sidx ne $sidx ) { - my ($ldapurl) = &get_ldap_url ($sidx, $sidx); + my ($ldapurl) = $supplierUrl = &get_ldap_url ($sidx, "n/a"); &print_legend if ( $last_sidx < 0); if($opt_s){ - print "Master: $ldapurl\n" + $sline = "Supplier: $ldapurl"; + $slen = length $sline; + print "\n$sline\n"; + for ($ii = 0; $ii < $slen; $ii++){ + print "-"; + } + print "\n"; } else { print "<p><p><hr><p>\n"; print "\n<p><center class=page-subtitle><font color=#0099cc>\n"; - print "Master:  $ldapurl</center>\n"; + print "Supplier:  $ldapurl</center>\n"; } $last_sidx = $sidx; + } else { + print "\n"; }
- # Print the current replica info onthe master + # Print the current replica info on the master if($opt_s){ - print "Replica ID: $serverid\n"; print "Replica Root: $replicaroot\n"; + print "Replica ID: $serverid\n"; print "Max CSN: $maxcsn\n"; } else { print "\n<p><table border=0 cellspacing=1 cellpadding=6 cols=10 width=100% class=bgColor9>\n"; @@ -702,11 +736,11 @@ sub print_hub_header $last_sidx = $sidx; }
- # Print the current replica info onthe master + # Print the current replica info on the master if($opt_s){ + print "\nReplica Root: $replicaroot\n"; print "Replica ID: $serverid\n"; - print "Replica Root: $replicaroot\n"; - print "Max CSN: $maxcsn\n"; + print "Max CSN: $maxcsn\n-\n"; } else { print "\n<p><table border=0 cellspacing=1 cellpadding=6 cols=10 width=100% class=bgColor9>\n"; print "\n<tr><td colspan=10><center>\n"; @@ -726,9 +760,10 @@ sub print_consumer_header #Print the header of consumer print "\n<tr class=bgColor16>\n"; - print "<th nowrap>Receiver</th>\n"; + print "<th nowrap>Consumer</th>\n"; print "<th nowrap>Time Lag</th>\n"; - print "<th nowrap>Max CSN</th>\n"; + print "<th nowrap>Supplier Max CSN</th>\n"; + print "<th nowrap>Consumer Max CSN</th>\n"; print "<th nowrap>Last Modify Time</th>\n"; print "<th nowrap>Supplier</th>\n"; print "<th nowrap>Sent/Skipped</th>\n"; @@ -749,6 +784,7 @@ sub print_consumers my ($c_maxcsn, $c_maxcsn_str, $c_lastmodified, $c_sidx, $lag, $markcolor); my ($c_replicaroot, $c_replicatype); my ($first_entry, $s_ldapurl, $c_ldapurl); + my $supplier_maxcsn = "_"; my ($nrows); my ($found);
@@ -781,12 +817,10 @@ sub print_consumers foreach (@consumers) { $c_ridx = $_; next if $c_ridx == $m_ridx; - + if ($c_ridx >= 0) { $myruv = $allruvs {"$c_ridx:$mid"}; ($c_maxcsn, $c_lastmodified) = split ( /;/, "$myruv" ); - ($c_maxcsn_str, $lag, $markcolor) = &cacl_time_lag ($m_maxcsn, $c_maxcsn); - if(!$opt_s){ $c_maxcsn_str =~ s/ /<br>/; } ($c_sidx, $c_replicaroot, $c_replicatype) = split (/:/, $allreplicas[$c_ridx]); $c_replicaroot = "same as master" if $m_replicaroot eq $c_replicaroot; } @@ -802,33 +836,49 @@ sub print_consumers
$nrows = 0; foreach (@ouragreements) { - next if ($_->{cidx} != $c_ridx); + $s_ridx = $_->{ridx}; + $s_sidx = $1 if $allreplicas [$s_ridx] =~ /^(\d+):/; + $s_ldapurl = &get_ldap_url ($s_sidx, "n/a"); + next if ($_->{cidx} != $c_ridx || $supplierUrl ne $s_ldapurl); $nrows++; }
$first_entry = 1; foreach (@ouragreements) { - next if ($_->{cidx} != $c_ridx); $s_ridx = $_->{ridx}; + $s_sidx = $1 if $allreplicas [$s_ridx] =~ /^(\d+):/; + $s_ldapurl = &get_ldap_url ($s_sidx, "n/a"); + next if ($_->{cidx} != $c_ridx || $supplierUrl ne $s_ldapurl); $conntype = $_->{nsDS5ReplicaBindMethod}; $status = $_->{nsds5replicaLastUpdateStatus}; $schedule = $_->{nsds5ReplicaUpdateSchedule}; - $s_sidx = $1 if $allreplicas [$s_ridx] =~ /^(\d+):/; - $s_ldapurl = &get_ldap_url ($s_sidx, "n/a");
# Print out the consumer's replica and ruvs if(!$opt_s){ print "\n<tr class=bgColor13>\n"; } if ($first_entry) { $first_entry = 0; $c_ldapurl = &get_ldap_url ($c_sidx, $conntype); + if ($c_ridx >= 0) { + ($c_maxcsn_str, $lag, $markcolor, $supplier_maxcsn) = + &cacl_time_lag ($_->{nsDS5ReplicaRoot}, + $_->{cn}, + $_->{nsds5ReplicaHost}, + $_->{nsds5ReplicaPort}, + $s_ridx, + $m_maxcsn, + $c_maxcsn); + if(!$opt_s){ $c_maxcsn_str =~ s/ /<br>/; } + } if($opt_s){ - print "Receiver: $c_ldapurl\nType: $c_replicatype\n"; + print "-\nConsumer: $c_ldapurl\nType: $c_replicatype\n"; print "Time Lag: $lag\n"; - print "Max CSN: $c_maxcsn_str\n"; + print "Supplier Max CSN: $supplier_maxcsn\n"; + print "Consumer Max CSN: $c_maxcsn_str\n"; print "Last Modify Time: $c_lastmodified\n"; } else { print "<td rowspan=$nrows width=5% class=bgColor5>$c_ldapurl<BR>Type: $c_replicatype</td>\n"; print "<td rowspan=$nrows width=5% nowrap bgcolor=$markcolor><center>$lag</center></td>\n"; + print "<td rowspan=$nrows width=15% nowrap>$supplier_maxcsn</td>\n"; print "<td rowspan=$nrows width=15% nowrap>$c_maxcsn_str</td>\n"; print "<td rowspan=$nrows width=15% nowrap>$c_lastmodified</td>\n"; } @@ -908,15 +958,44 @@ sub print_consumers } }
+sub get_supplier_maxcsn +{ + my ($ridx, $s, $cn, $h, $p) = @_; + my $decimalcsn; + my $csn = ""; + # normalize suffix + $s =~ s/ //; + $s =~ lc $s; + my $agmt = "$ridx;$s;$cn;$h;$p"; + + foreach my $key (keys %agmtmaxcsn){ + if ($key eq $agmt){ + $csn = $agmtmaxcsn{$key}; + last; + } + } + if($csn ne ""){ + $decimalcsn = &to_decimal_csn ($csn); + return "$csn:$decimalcsn"; + } + + return $csn; +} + sub cacl_time_lag { - my ($s_maxcsn, $c_maxcsn) = @_; + my ($s, $cn, $h, $p, $ridx, $s_maxcsn, $c_maxcsn) = @_; my ($markcolor); - my ($csn_str); + my ($m_maxcsn, $csn_str, $supplier_csn_str); my ($s_tm, $c_tm, $lag_tm, $lag_str, $hours, $minutes);
+ $m_maxcsn = get_supplier_maxcsn($ridx,$s, $cn, $h, $p); + if($m_maxcsn ne ""){ + $s_maxcsn = $m_maxcsn; + } + $supplier_csn_str = &to_string_csn ($s_maxcsn); $csn_str = &to_string_csn ($c_maxcsn); - + if ($s_maxcsn && !$c_maxcsn) { $lag_str = "- ?:??:??"; $markcolor = &get_color (36000); # assume consumer has big latency @@ -957,7 +1036,33 @@ sub cacl_time_lag $lag_tm = "0".$lag_tm if ($lag_tm < 10); $lag_str .= "$lag_tm"; } - return ($csn_str, $lag_str, $markcolor); + return ($csn_str, $lag_str, $markcolor, $supplier_csn_str); +} + +sub set_server_params +{ + my ($host, $port, $binddn, $bindpwd, $bindcert); + + ($host, $port, $binddn, $bindpwd, $bindcert) = split (/:/, $allconnections[0]); + if($opt_p && $opt_p ne ""){ + $ld{port} = $opt_p + } elsif(!$port || $port eq ""){ + $ld{port} = "389"; + } else { + $ld{port} = $port; + } + if($host && $host ne ""){ + $ld{host} = $host; + } + if($binddn){ + $ld{bind} = $binddn; + } + if($bindpwd){ + $ld{pswd} = $bindpwd; + } + if($bindcert){ + $ld{cert} = $bindcert; + } }
# @@ -1005,24 +1110,23 @@ sub add_server $c = ""; } if(!$w || $w eq "*"){ - $w = ""; - } + $w = ""; + } } if (!$binddn || $binddn eq "" || $binddn eq "*") { if ($d eq "" || $d eq "*") { $binddn = "cn=Directory Manager"; - } - else { + } else { $binddn = $d; } } if($prompt eq "yes" && ($w eq "" || (!$bindpwd || $bindpwd eq ""))){ - $bindpwd = passwdPrompt($h, $p); + $bindpwd = passwdPrompt($h, $p); } elsif ($passwd ne ""){ - $bindpwd = $passwd; + $bindpwd = $passwd; } else { - $bindpwd = $w if (!$bindpwd || $bindpwd eq "" || $bindpwd eq "*"); - } + $bindpwd = $w if (!$bindpwd || $bindpwd eq "" || $bindpwd eq "*"); + } $bindcert = $c if (!$bindcert || $bindcert eq "" || $bindcert eq "*"); }
@@ -1037,38 +1141,38 @@ sub add_server sub passwdPrompt { - my ($h, $p) = @_; - my $key = "$h:$p"; - my $pw = ""; + my ($h, $p) = @_; + my $key = "$h:$p"; + my $pw = "";
- if ($passwords{$key}){ - # we already have a password for this replica - return $passwords{$key}; - } - # Disable console echo - system("@sttyexec@ -echo") if -t STDIN; - - while ($pw eq ""){ - if($passwd ne ""){ - print "Enter password for ($h:$p) <hit Enter to use previous password>: "; - chomp($pw = <>); - if ($pw eq ""){ - $pw = $passwd; - } else { - $passwords{$key} = $pw; - $passwd = $pw; - } - } else { - print "Enter password for ($h:$p): "; - chomp($pw = <>); - $passwords{$key} = $pw; - $passwd = $pw; - } - } - # Enable console echo - system("@sttyexec@ echo") if -t STDIN; + if ($passwords{$key}){ + # we already have a password for this replica + return $passwords{$key}; + } + # Disable console echo + system("@sttyexec@ -echo") if -t STDIN; + + while ($pw eq ""){ + if($passwd ne ""){ + print "Enter password for ($h:$p) <hit Enter to use previous password>: "; + chomp($pw = <>); + if ($pw eq ""){ + $pw = $passwd; + } else { + $passwords{$key} = $pw; + $passwd = $pw; + } + } else { + print "Enter password for ($h:$p): "; + chomp($pw = <>); + $passwords{$key} = $pw; + $passwd = $pw; + } + } + # Enable console echo + system("@sttyexec@ echo") if -t STDIN;
- return $pw; + return $pw; }
sub get_ldap_url @@ -1131,6 +1235,7 @@ sub to_string_csn } my ($csnstr) = "$mon/$mday/$year $hour:$min:$sec"; $csnstr .= " $seq $subseq" if ( $seq != 0 || $subseq != 0 ); + return "$rawcsn ($csnstr)"; }
@@ -1207,7 +1312,7 @@ sub print_html_header if ($opt_u) { print " - This report updates every $interval seconds\n\n"; } else { - print "\n\n"; + print "\n"; } } } diff --git a/ldap/servers/plugins/replication/repl5.h b/ldap/servers/plugins/replication/repl5.h index 19fed0c..eb646e3 100644 --- a/ldap/servers/plugins/replication/repl5.h +++ b/ldap/servers/plugins/replication/repl5.h @@ -199,6 +199,7 @@ extern const char *attr_flags; extern const char *attr_replicaName; extern const char *attr_replicaReferral; extern const char *type_ruvElement; +extern const char *type_agmtMaxCSN; extern const char *type_replicaPurgeDelay; extern const char *type_replicaChangeCount; extern const char *type_replicaTombstonePurgeInterval; @@ -310,6 +311,8 @@ typedef struct repl_bos Repl_Bos;
/* In repl5_agmt.c */ typedef struct repl5agmt Repl_Agmt; +typedef struct replica Replica; + #define TRANSPORT_FLAG_SSL 1 #define TRANSPORT_FLAG_TLS 2 #define BINDMETHOD_SIMPLE_AUTH 1 @@ -391,8 +394,11 @@ int agmt_set_attrs_to_strip(Repl_Agmt *ra, Slapi_Entry *e); int agmt_set_timeout(Repl_Agmt *ra, long timeout); void agmt_update_done(Repl_Agmt *ra, int is_total); int agmt_get_protocol_timeout(Repl_Agmt *agmt); - -typedef struct replica Replica; +void agmt_update_maxcsn(Replica *r, Slapi_DN *sdn, int op, LDAPMod **mods, CSN *csn); +void add_agmt_maxcsns(Slapi_Entry *e, Replica *r); +void agmt_set_maxcsn(Repl_Agmt *ra); +void agmt_remove_maxcsn(Repl_Agmt *ra); +int agmt_maxcsn_to_smod (Replica *r, Slapi_Mod *smod);
/* In repl5_agmtlist.c */ int agmtlist_config_init(); @@ -601,6 +607,9 @@ int replica_get_backoff_min(Replica *r); int replica_get_backoff_max(Replica *r); void replica_set_backoff_min(Replica *r, int min); void replica_set_backoff_max(Replica *r, int max); +int replica_get_agmt_count(Replica *r); +void replica_incr_agmt_count(Replica *r); +void replica_decr_agmt_count(Replica *r);
/* The functions below handles the state flag */ /* Current internal state flags */ diff --git a/ldap/servers/plugins/replication/repl5_agmt.c b/ldap/servers/plugins/replication/repl5_agmt.c index d3ee568..d78f675 100644 --- a/ldap/servers/plugins/replication/repl5_agmt.c +++ b/ldap/servers/plugins/replication/repl5_agmt.c @@ -125,6 +125,7 @@ typedef struct repl5agmt { Object *consumerRUV; /* last RUV received from the consumer - used for changelog purging */ CSN *consumerSchemaCSN; /* last schema CSN received from the consumer */ ReplicaId consumerRID; /* indicates if the consumer is the originator of a CSN */ + int tmpConsumerRID; /* Indicates the consumer rid was set from the agmt maxcsn - it should be refreshed */ long timeout; /* timeout (in seconds) for outbound LDAP connections to remote server */ PRBool stop_in_progress; /* set by agmt_stop when shutting down */ long busywaittime; /* time in seconds to wait after getting a REPLICA BUSY from the consumer - @@ -142,6 +143,8 @@ typedef struct repl5agmt { * modifiersname, modifytimestamp, internalModifiersname, internalModifyTimestamp, etc */ int agreement_type; PRUint64 protocol_timeout; + char *maxcsn; /* agmt max csn */ + Slapi_RWLock *attr_lock; /* RW lock for all the stripped attrs */ } repl5agmt;
/* Forward declarations */ @@ -151,7 +154,7 @@ static int get_agmt_status(Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter, int *returncode, char *returntext, void *arg); static int agmt_set_bind_method_no_lock(Repl_Agmt *ra, const Slapi_Entry *e); static int agmt_set_transportinfo_no_lock(Repl_Agmt *ra, const Slapi_Entry *e); - +static ReplicaId agmt_maxcsn_get_rid(char *maxcsn);
/* Schema for replication agreement: @@ -264,6 +267,13 @@ agmt_new_from_entry(Slapi_Entry *e) slapi_entry_get_dn_const(e)); goto loser; } + if ((ra->attr_lock = slapi_new_rwlock()) == NULL) + { + slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Unable to create new attr lock " + "for replication agreement "%s" - agreement ignored.\n", + slapi_entry_get_dn_const(e)); + goto loser; + }
/* Find all the stuff we need for the agreement */
@@ -545,6 +555,8 @@ void agmt_delete(void **rap) { Repl_Agmt *ra; + Replica *replica = NULL; + Object *repl_obj = NULL; PR_ASSERT(NULL != rap); PR_ASSERT(NULL != *rap);
@@ -565,7 +577,7 @@ agmt_delete(void **rap) LDAP_SCOPE_BASE, "(objectclass=*)", get_agmt_status);
- /* + /* * Call the replication session cleanup callback. We * need to do this before we free replarea. */ @@ -573,23 +585,31 @@ agmt_delete(void **rap) repl_session_plugin_call_destroy_agmt_cb(ra); }
- /* slapi_ch_free accepts NULL pointer */ - slapi_ch_free((void **)&(ra->hostname)); - slapi_ch_free((void **)&(ra->binddn)); - + slapi_ch_free_string(&ra->hostname); + slapi_ch_free_string(&ra->binddn); slapi_ch_array_free(ra->frac_attrs); slapi_ch_array_free(ra->frac_attrs_total);
if (NULL != ra->creds) { - /* XXX free berval */ + ber_bvfree(ra->creds); } if (NULL != ra->replarea) { + /* + * Get the replica for this agreement from the repl area + * so we can decrement the agmt count + */ + repl_obj = replica_get_replica_from_dn(ra->replarea); + if (repl_obj) { + replica = (Replica*)object_get_data (repl_obj); + replica_decr_agmt_count(replica); + object_release(repl_obj); + } slapi_sdn_free(&ra->replarea); }
- if (NULL != ra->consumerRUV) + if (NULL != ra->consumerRUV) { object_release (ra->consumerRUV); } @@ -597,20 +617,26 @@ agmt_delete(void **rap) csn_free (&ra->consumerSchemaCSN); while ( --(ra->num_changecounters) >= 0 ) { - slapi_ch_free((void **)&ra->changecounters[ra->num_changecounters]); + slapi_ch_free((void **)&ra->changecounters[ra->num_changecounters]); }
if (ra->agreement_type == REPLICA_TYPE_WINDOWS) { windows_agreement_delete(ra); } - if(ra->attrs_to_strip){ slapi_ch_array_free(ra->attrs_to_strip); } - + if(ra->maxcsn){ + slapi_ch_free_string(&ra->maxcsn); + } schedule_destroy(ra->schedule); - slapi_ch_free((void **)&ra->long_name); + slapi_ch_free_string(&ra->long_name); + + /* free the locks */ + PR_DestroyLock(ra->lock); + slapi_destroy_rwlock(ra->attr_lock); + slapi_ch_free((void **)rap); }
@@ -658,6 +684,8 @@ agmt_start(Repl_Agmt *ra) /* Start the protocol thread */ prot_start(ra->protocol);
+ agmt_set_maxcsn(ra); + PR_Unlock(ra->lock); return 0; } @@ -876,9 +904,9 @@ agmt_get_fractional_attrs(const Repl_Agmt *ra) { return NULL; } - PR_Lock(ra->lock); + slapi_rwlock_rdlock(ra->attr_lock); return_value = charray_dup(ra->frac_attrs); - PR_Unlock(ra->lock); + slapi_rwlock_unlock(ra->attr_lock); return return_value; }
@@ -907,10 +935,10 @@ agmt_is_fractional_attr(const Repl_Agmt *ra, const char *attrname) { return 0; } - PR_Lock(ra->lock); + slapi_rwlock_rdlock(ra->attr_lock); /* Scan the list looking for a match */ return_value = charray_inlist(ra->frac_attrs,(char*)attrname); - PR_Unlock(ra->lock); + slapi_rwlock_unlock(ra->attr_lock); return return_value; }
@@ -1355,7 +1383,7 @@ agmt_set_replicated_attributes_from_entry(Repl_Agmt *ra, const Slapi_Entry *e)
PR_ASSERT(NULL != ra); slapi_entry_attr_find(e, type_nsds5ReplicatedAttributeList, &sattr); - PR_Lock(ra->lock); + slapi_rwlock_wrlock(ra->attr_lock); if (ra->frac_attrs) { slapi_ch_array_free(ra->frac_attrs); @@ -1372,7 +1400,7 @@ agmt_set_replicated_attributes_from_entry(Repl_Agmt *ra, const Slapi_Entry *e) return_value = agmt_parse_excluded_attrs_config_attr(val,&(ra->frac_attrs)); } } - PR_Unlock(ra->lock); + slapi_rwlock_unlock(ra->attr_lock); prot_notify_agmt_changed(ra->protocol, ra->long_name); return return_value; } @@ -1422,7 +1450,7 @@ agmt_set_replicated_attributes_from_attr(Repl_Agmt *ra, Slapi_Attr *sattr) int return_value = 0;
PR_ASSERT(NULL != ra); - PR_Lock(ra->lock); + slapi_rwlock_wrlock(ra->attr_lock); if (ra->frac_attrs) { slapi_ch_array_free(ra->frac_attrs); @@ -1439,7 +1467,7 @@ agmt_set_replicated_attributes_from_attr(Repl_Agmt *ra, Slapi_Attr *sattr) return_value = agmt_parse_excluded_attrs_config_attr(val,&(ra->frac_attrs)); } } - PR_Unlock(ra->lock); + slapi_rwlock_unlock(ra->attr_lock); return return_value; }
@@ -1847,6 +1875,7 @@ agmt_notify_change(Repl_Agmt *agmt, Slapi_PBlock *pb) int i, j;
slapi_pblock_get(pb, SLAPI_MODIFY_MODS, &mods); + slapi_rwlock_rdlock(agmt->attr_lock); for (i = 0; !affects_non_fractional_attribute && NULL != agmt->frac_attrs[i]; i++) { for (j = 0; !affects_non_fractional_attribute && NULL != mods[j]; j++) @@ -1858,6 +1887,7 @@ agmt_notify_change(Repl_Agmt *agmt, Slapi_PBlock *pb) } } } + slapi_rwlock_unlock(agmt->attr_lock); } else { @@ -2454,12 +2484,11 @@ update_window_state_change_callback (void *arg, PRBool opened) ReplicaId agmt_get_consumer_rid ( Repl_Agmt *agmt, void *conn ) { - if ( agmt->consumerRID <= 0 ) { + if ( agmt->consumerRID <= 0 || agmt->tmpConsumerRID) {
char *mapping_tree_node = NULL; struct berval **bvals = NULL;
- /* This function converts the old style DN to the new one. */ mapping_tree_node = slapi_create_dn_string("cn=replica,cn="%s",cn=mapping tree,cn=config", @@ -2482,6 +2511,7 @@ agmt_get_consumer_rid ( Repl_Agmt *agmt, void *conn ) } slapi_ch_free_string(&mapping_tree_node); } + agmt->tmpConsumerRID = 0;
return agmt->consumerRID; } @@ -2661,3 +2691,401 @@ agmt_get_protocol_timeout(Repl_Agmt *agmt) return (int)agmt->protocol_timeout; }
+/* + * Check if all the mods are being removed by fractional/stripped attributes + */ +void +agmt_update_maxcsn(Replica *r, Slapi_DN *sdn, int op, LDAPMod **mods, CSN *csn) +{ + Object *agmt_obj; + Repl_Agmt *agmt; + ReplicaId rid = replica_get_rid(r); + int mod_count = 0, excluded_count = 0; + + agmt_obj = agmtlist_get_first_agreement_for_replica (r); + if(agmt_obj == NULL){ /* no agreements */ + return; + } + while (agmt_obj){ + agmt = (Repl_Agmt*)object_get_data (agmt_obj); + if(!agmt_is_enabled(agmt) || + !slapi_sdn_issuffix(sdn, agmt->replarea) || + get_agmt_agreement_type(agmt) == REPLICA_TYPE_WINDOWS) + { + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + continue; + } + + if(op == SLAPI_OPERATION_MODIFY) + { + slapi_rwlock_rdlock(agmt->attr_lock); + for ( excluded_count = 0, mod_count = 0; NULL != mods[ mod_count ]; mod_count++){ + if(charray_inlist(agmt->frac_attrs, mods[mod_count]->mod_type)){ + excluded_count++; + } else if(charray_inlist(agmt->attrs_to_strip, mods[mod_count]->mod_type)){ + excluded_count++; + } + } + slapi_rwlock_unlock(agmt->attr_lock); + } + + if(excluded_count == 0 || excluded_count != mod_count){ + /* + * This update has not been completely stripped down, update + * the agmt maxcsn - if the update did not originate from the consumer. + */ + char maxcsn[CSN_STRSIZE]; + ReplicaId oprid = csn_get_replicaid(csn); + + csn_as_string(csn, PR_FALSE, maxcsn); + PR_Lock(agmt->lock); + if(!agmt->consumerRID){ + /* + * If the RID is 0, that means this is the first update since the + * agmt was created. Since we have not contacted the consumer yet, + * we don't know what its rid is. The consumerRID will be set once + * this update is sent, but until then we don't know it. So for now + * temporarily mark it as "unavailable". + */ + slapi_ch_free_string(&agmt->maxcsn); + agmt->maxcsn = slapi_ch_smprintf("%s;%s;%s;%d;unavailable", slapi_sdn_get_dn(agmt->replarea), + slapi_rdn_get_value_by_ref(slapi_rdn_get_rdn(agmt->rdn)), agmt->hostname, agmt->port); + } else if(rid == oprid){ + slapi_ch_free_string(&agmt->maxcsn); + agmt->maxcsn = slapi_ch_smprintf("%s;%s;%s;%d;%d;%s", slapi_sdn_get_dn(agmt->replarea), + slapi_rdn_get_value_by_ref(slapi_rdn_get_rdn(agmt->rdn)), agmt->hostname, + agmt->port, agmt->consumerRID, maxcsn); + } + PR_Unlock(agmt->lock); + } + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + } +} + +/* + * Returns the in-memory agmt maxcsn's + */ +void +add_agmt_maxcsns(Slapi_Entry *e, Replica *r) +{ + Object *agmt_obj; + Repl_Agmt *agmt; + + agmt_obj = agmtlist_get_first_agreement_for_replica (r); + if(agmt_obj == NULL){ /* no agreements */ + return; + } + while (agmt_obj){ + agmt = (Repl_Agmt*)object_get_data (agmt_obj); + if(!agmt_is_enabled(agmt) || get_agmt_agreement_type(agmt) == REPLICA_TYPE_WINDOWS){ + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + continue; + } + PR_Lock(agmt->lock); + if(agmt->maxcsn){ + slapi_entry_add_string(e, type_agmtMaxCSN, agmt->maxcsn); + } + PR_Unlock(agmt->lock); + + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + } +} + +/* + * Create a smod of all the agmt maxcsns to add to the tombstone entry + */ +int +agmt_maxcsn_to_smod (Replica *r, Slapi_Mod *smod) +{ + Object *agmt_obj; + Repl_Agmt *agmt; + int rc = 1; + + agmt_obj = agmtlist_get_first_agreement_for_replica (r); + if(agmt_obj == NULL){ /* no agreements */ + return rc; + } + slapi_mod_init (smod, replica_get_agmt_count(r) + 1); + slapi_mod_set_type (smod, type_agmtMaxCSN); + slapi_mod_set_operation (smod, LDAP_MOD_REPLACE | LDAP_MOD_BVALUES); + + while (agmt_obj){ + struct berval val; + + agmt = (Repl_Agmt*)object_get_data (agmt_obj); + if(!agmt_is_enabled(agmt) || get_agmt_agreement_type(agmt) == REPLICA_TYPE_WINDOWS){ + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + continue; + } + PR_Lock(agmt->lock); + if(agmt->maxcsn == NULL){ + PR_Unlock(agmt->lock); + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + continue; + } + val.bv_val = agmt->maxcsn; + val.bv_len = strlen(val.bv_val); + slapi_mod_add_value(smod, &val); + PR_Unlock(agmt->lock); + rc = 0; + + agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj); + } + return rc; +} + +/* + * Called when we start a repl agmt + */ +void +agmt_set_maxcsn(Repl_Agmt *ra) +{ + Slapi_PBlock *pb = NULL; + Slapi_Entry **entries = NULL; + Replica *r = NULL; + Object *repl_obj; + const Slapi_DN *tombstone_sdn = NULL; + char *attrs[2]; + int rc; + + /* read ruv state from the ruv tombstone entry */ + pb = slapi_pblock_new(); + if (!pb) { + slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "agmt_set_maxcsn: Out of memory\n"); + goto done; + } + repl_obj = prot_get_replica_object(ra->protocol); + if(repl_obj){ + r = (Replica *)object_get_data(repl_obj); + tombstone_sdn = replica_get_root(r); + } + ra->maxcsn = NULL; + attrs[0] = (char*)type_agmtMaxCSN; + attrs[1] = NULL; + slapi_search_internal_set_pb_ext( + pb, + (Slapi_DN *)tombstone_sdn, + LDAP_SCOPE_BASE, + "objectclass=*", + attrs, + 0, /* attrsonly */ + NULL, /* controls */ + RUV_STORAGE_ENTRY_UNIQUEID, + repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), + OP_FLAG_REPLICATED); /* flags */ + slapi_search_internal_pb (pb); + + slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc); + if (rc == LDAP_SUCCESS){ + ReplicaId rid; + Replica *r; + Object *repl_obj; + char **maxcsns; + int i; + + slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &entries); + if (NULL == entries || NULL == entries[0]){ + slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, + "agmt_set_maxcsn: replica ruv tombstone entry for " + "replica %s not found\n", + slapi_sdn_get_dn(ra->replarea)); + goto done; + } + maxcsns = slapi_entry_attr_get_charray(entries[0], type_agmtMaxCSN); + repl_obj = prot_get_replica_object(ra->protocol); + if(repl_obj && maxcsns){ + r = (Replica *)object_get_data(repl_obj); + if(r){ + rid = replica_get_rid(r); + /* + * Loop over all the agmt maxcsns and find ours + */ + for(i = 0; maxcsns[i]; i++){ + char buf[BUFSIZ]; + char unavail_buf[BUFSIZ]; + + PR_snprintf(buf,BUFSIZ,"%s;%s;%s;%d;",slapi_sdn_get_dn(ra->replarea), + slapi_rdn_get_value_by_ref(slapi_rdn_get_rdn(ra->rdn)), + ra->hostname, ra->port); + PR_snprintf(unavail_buf, BUFSIZ,"%s;%s;%s;%d;unavailable", slapi_sdn_get_dn(ra->replarea), + slapi_rdn_get_value_by_ref(slapi_rdn_get_rdn(ra->rdn)), + ra->hostname, ra->port); + if(strstr(maxcsns[i], buf) || strstr(maxcsns[i], unavail_buf)){ + ra->maxcsn = slapi_ch_strdup(maxcsns[i]); + ra->consumerRID = agmt_maxcsn_get_rid(maxcsns[i]); + ra->tmpConsumerRID = 1; + break; + } + } + } + } + slapi_ch_array_free(maxcsns); + } +done: + if (NULL != pb){ + slapi_free_search_results_internal(pb); + slapi_pblock_destroy (pb); + } +} + +/* + * Parse out the consumer replicaID from the agmt maxcsn + * + * "repl area;agmt rdn;hostname;port;consumer rid;maxcsn" + */ +static ReplicaId +agmt_maxcsn_get_rid(char *maxcsn) +{ + ReplicaId rid = 0; + char *token = NULL; + char *iter; + char *value = slapi_ch_strdup(maxcsn); + + token = ldap_utf8strtok_r(value, ";", &iter); /* repl area */ + token = ldap_utf8strtok_r(iter, ";", &iter); /* agmt rdn */ + token = ldap_utf8strtok_r(iter, ";", &iter); /* host */ + token = ldap_utf8strtok_r(iter, ";", &iter); /* port */ + token = ldap_utf8strtok_r(iter, ";", &iter); /* rid */ + + if(strcmp(token, "Unavailable")){ + rid = atoi(token); + } + slapi_ch_free_string(&value); + + return rid; +} + +/* + * Agmt being deleted, remove the agmt maxcsn from the local ruv. + */ +void +agmt_remove_maxcsn(Repl_Agmt *ra) +{ + Slapi_PBlock *pb = NULL; + Slapi_PBlock *modpb = NULL; + Slapi_Entry **entries = NULL; + Replica *r = NULL; + Object *repl_obj; + const Slapi_DN *tombstone_sdn = NULL; + char *attrs[2]; + int rc; + + pb = slapi_pblock_new(); + if (!pb) { + slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "agmt_set_maxcsn: Out of memory\n"); + goto done; + } + ra->maxcsn = NULL; + + repl_obj = prot_get_replica_object(ra->protocol); + if(repl_obj){ + r = (Replica *)object_get_data(repl_obj); + tombstone_sdn = replica_get_root(r); + } else { + slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "agmt_set_maxcsn: Failed to get repl object.\n"); + goto done; + } + attrs[0] = (char*)type_agmtMaxCSN; + attrs[1] = NULL; + + slapi_search_internal_set_pb_ext( + pb, + (Slapi_DN *)tombstone_sdn, + LDAP_SCOPE_BASE, + "objectclass=*", + attrs, + 0, /* attrsonly */ + NULL, /* controls */ + RUV_STORAGE_ENTRY_UNIQUEID, + repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), + OP_FLAG_REPLICATED); /* flags */ + slapi_search_internal_pb (pb); + slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc); + + if (rc == LDAP_SUCCESS){ + /* + * Ok we have the db tombstone entry, start looking through the agmt maxcsns + * for a match to this replica agmt. + */ + ReplicaId rid; + Slapi_Mod smod; + LDAPMod *mods[2]; + char **maxcsns = NULL; + int i; + + slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &entries); + if (NULL == entries || NULL == entries[0]){ + slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, + "agmt_set_maxcsn: replica ruv tombstone entry for " + "replica %s not found\n", slapi_sdn_get_dn(ra->replarea)); + goto done; + } + maxcsns = slapi_entry_attr_get_charray(entries[0], type_agmtMaxCSN); + repl_obj = prot_get_replica_object(ra->protocol); + if(repl_obj && maxcsns){ + r = (Replica *)object_get_data(repl_obj); + if(r){ + rid = replica_get_rid(r); + /* + * Loop over all the agmt maxcsns and find ours... + */ + for(i = 0; maxcsns[i]; i++){ + char buf[BUFSIZ]; + char unavail_buf[BUFSIZ]; + char *val; + + PR_snprintf(buf, BUFSIZ,"%s;%s;%s;%d;",slapi_sdn_get_dn(ra->replarea), + slapi_rdn_get_value_by_ref(slapi_rdn_get_rdn(ra->rdn)), + ra->hostname, ra->port); + PR_snprintf(unavail_buf, BUFSIZ, "%s;%s;%s;%d;unavailable", + slapi_sdn_get_dn(ra->replarea), + slapi_rdn_get_value_by_ref(slapi_rdn_get_rdn(ra->rdn)), + ra->hostname, ra->port); + if(strstr(maxcsns[i], buf) || strstr(maxcsns[i], unavail_buf)){ + /* + * We found the matching agmt maxcsn, now remove agmt maxcsn + * from the tombstone entry. + */ + val = maxcsns[i]; + modpb = slapi_pblock_new(); + slapi_mod_init (&smod, 2); + slapi_mod_set_type (&smod, type_agmtMaxCSN); + slapi_mod_set_operation (&smod, LDAP_MOD_DELETE | LDAP_MOD_BVALUES); + mods [0] = smod.mod; + mods [1] = NULL; + modpb = slapi_pblock_new(); + + slapi_modify_internal_set_pb_ext( + modpb, + tombstone_sdn, + mods, + NULL, /* controls */ + RUV_STORAGE_ENTRY_UNIQUEID, + repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), + /* Add OP_FLAG_TOMBSTONE_ENTRY so that this doesn't get logged in the Retro ChangeLog */ + OP_FLAG_REPLICATED | OP_FLAG_REPL_FIXUP | OP_FLAG_TOMBSTONE_ENTRY | + OP_FLAG_REPL_RUV ); + slapi_modify_internal_pb(modpb); + slapi_pblock_get(modpb, SLAPI_PLUGIN_INTOP_RESULT, &rc); + if (rc != LDAP_SUCCESS){ + slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, + "agmt_set_maxcsn: failed to remove agmt maxcsn (%s), error(%d)\n",maxcsns[i], rc); + } + slapi_mod_done(&smod); + slapi_pblock_destroy(modpb); + break; + } + } + } + } + slapi_ch_array_free(maxcsns); + } + +done: + if (NULL != pb){ + slapi_free_search_results_internal(pb); + slapi_pblock_destroy (pb); + } +} + + diff --git a/ldap/servers/plugins/replication/repl5_agmtlist.c b/ldap/servers/plugins/replication/repl5_agmtlist.c index e4ef66b..01814b0 100644 --- a/ldap/servers/plugins/replication/repl5_agmtlist.c +++ b/ldap/servers/plugins/replication/repl5_agmtlist.c @@ -171,6 +171,7 @@ add_new_agreement(Slapi_Entry *e) }
rc = replica_start_agreement(replica, ra); + replica_incr_agmt_count(replica);
if (repl_obj) object_release(repl_obj);
@@ -396,6 +397,12 @@ agmtlist_modify_callback(Slapi_PBlock *pb, Slapi_Entry *entryBefore, Slapi_Entry agmt_get_long_name(agmt)); *returncode = LDAP_OPERATIONS_ERROR; rc = SLAPI_DSE_CALLBACK_ERROR; + } else { + /* + * Changing the port invalidates the agmt maxcsn, so remove it. + * The next update will add the correct maxcsn back to the agmt/local ruv. + */ + agmt_remove_maxcsn(agmt); } } else if (slapi_attr_types_equivalent(mods[i]->mod_type, @@ -586,6 +593,7 @@ agmtlist_delete_callback(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *entryAft } else { + agmt_remove_maxcsn(ra); /* remove the agmt maxcsn from the localruv */ agmt_stop(ra); object_release(ro); /* Release ref acquired in objset_find */ objset_remove_obj(agmt_set, ro); /* Releases a reference (should be final reference */ diff --git a/ldap/servers/plugins/replication/repl5_plugins.c b/ldap/servers/plugins/replication/repl5_plugins.c index 9c97ea6..43b9d2e 100644 --- a/ldap/servers/plugins/replication/repl5_plugins.c +++ b/ldap/servers/plugins/replication/repl5_plugins.c @@ -1206,12 +1206,19 @@ write_changelog_and_ruv (Slapi_PBlock *pb) CSN *opcsn; int rc; const char *dn = op_params ? REPL_GET_DN(&op_params->target_address) : "unknown"; + Slapi_DN *sdn = op_params ? (&op_params->target_address)->sdn : NULL; char *uniqueid = op_params ? op_params->target_address.uniqueid : "unknown"; unsigned long optype = op_params ? op_params->operation_type : 0; CSN *oppcsn = op_params ? op_params->csn : NULL; + LDAPMod **mods = op_params ? op_params->p.p_modify.modify_mods : NULL;
slapi_pblock_get( pb, SLAPI_OPERATION, &op ); opcsn = operation_get_csn(op); + + /* Update each agmt's maxcsn */ + if(op_params && sdn){ + agmt_update_maxcsn(r, sdn, op_params->operation_type, mods, opcsn); + } rc = update_ruv_component(r, opcsn, pb); if (RUV_COVERS_CSN == rc) { slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, diff --git a/ldap/servers/plugins/replication/repl5_protocol.c b/ldap/servers/plugins/replication/repl5_protocol.c index 34fe8a0..b56829c 100644 --- a/ldap/servers/plugins/replication/repl5_protocol.c +++ b/ldap/servers/plugins/replication/repl5_protocol.c @@ -72,7 +72,7 @@ typedef struct repl_protocol int state; int next_state; PRUint64 protocol_timeout; - PRThread *agmt_thread; + PRThread *agmt_thread; PRLock *lock; } repl_protocol;
diff --git a/ldap/servers/plugins/replication/repl5_replica.c b/ldap/servers/plugins/replication/repl5_replica.c index b07a9f9..d28b48c 100644 --- a/ldap/servers/plugins/replication/repl5_replica.c +++ b/ldap/servers/plugins/replication/repl5_replica.c @@ -90,6 +90,7 @@ struct replica { PRUint64 protocol_timeout; /* protocol shutdown timeout */ PRUint64 backoff_min; /* backoff retry minimum */ PRUint64 backoff_max; /* backoff retry maximum */ + PRUint64 agmt_count; /* Number of agmts */ };
@@ -609,7 +610,7 @@ replica_get_ruv (const Replica *r) return ruv; }
-/* +/* * Sets RUV vector. This function should be called during replica * (re)initialization. During normal operation, the RUV is read from * the root of the replicated in the replica_new call @@ -1629,12 +1630,13 @@ _replica_get_config_entry (const Slapi_DN *root, const char **attrs) Slapi_Entry * get_in_memory_ruv(Slapi_DN *suffix_sdn) { - const char *attrs[3]; + const char *attrs[4];
/* these two attributes needs to be asked when reading the RUV */ attrs[0] = type_ruvElement; attrs[1] = type_ruvElementUpdatetime; - attrs[2] = NULL; + attrs[2] = type_agmtMaxCSN; + attrs[3] = NULL; return(_replica_get_config_entry(suffix_sdn, attrs)); }
@@ -2224,7 +2226,6 @@ _replica_configure_ruv (Replica *r, PRBool isLocked) if (NULL != generation) { r->repl_ruv = object_new((void*)ruv, (FNFree)ruv_destroy); - /* Is the local purl in the ruv? (the port or the host could have changed) */ @@ -2579,10 +2580,11 @@ int replica_write_ruv (Replica *r) { int rc = LDAP_SUCCESS; - Slapi_Mod smod; + Slapi_Mod smod, rmod; Slapi_Mod smod_last_modified; - LDAPMod *mods [3]; + LDAPMod *mods [4]; Slapi_PBlock *pb; + int free_rmod = 0;
PR_ASSERT(r);
@@ -2603,7 +2605,13 @@ replica_write_ruv (Replica *r)
mods [0] = (LDAPMod *)slapi_mod_get_ldapmod_byref(&smod); mods [1] = (LDAPMod *)slapi_mod_get_ldapmod_byref(&smod_last_modified); - mods [2] = NULL; + if(agmt_maxcsn_to_smod(r,&rmod) == LDAP_SUCCESS){ + mods [2] = (LDAPMod *)slapi_mod_get_ldapmod_byref(&rmod); + free_rmod = 1; + } else { + mods [2] = NULL; + } + mods [3] = NULL; pb = slapi_pblock_new();
/* replica name never changes so it is ok to reference it outside the lock */ @@ -2625,7 +2633,7 @@ replica_write_ruv (Replica *r)
if (rc == LDAP_SUCCESS) { - r->repl_ruv_dirty = PR_FALSE; + r->repl_ruv_dirty = PR_FALSE; } else if (rc == LDAP_NO_SUCH_OBJECT) { @@ -2646,6 +2654,7 @@ replica_write_ruv (Replica *r) PR_Unlock(r->repl_lock); slapi_mod_done (&smod); + if(free_rmod) slapi_mod_done (&rmod); slapi_mod_done (&smod_last_modified); slapi_pblock_destroy (pb);
@@ -3928,3 +3937,21 @@ replica_set_backoff_max(Replica *r, int max) { r->backoff_max = max; } + +int +replica_get_agmt_count(Replica *r) +{ + return r->agmt_count; +} + +void +replica_incr_agmt_count(Replica *r) +{ + r->agmt_count++; +} + +void +replica_decr_agmt_count(Replica *r) +{ + r->agmt_count--; +} diff --git a/ldap/servers/plugins/replication/repl5_replica_config.c b/ldap/servers/plugins/replication/repl5_replica_config.c index a4d9e16..ea23837 100644 --- a/ldap/servers/plugins/replication/repl5_replica_config.c +++ b/ldap/servers/plugins/replication/repl5_replica_config.c @@ -740,9 +740,10 @@ replica_config_search_ruv(Slapi_PBlock *pb, Slapi_Entry* e, Replica *replica) slapi_entry_add_values_sv(e, type_ruvElement, values); valuearray_free(&values); } - object_release(ruv_obj);
+ /* now add all the repl agmts maxcsnsruv */ + add_agmt_maxcsns(e, replica); }
/* Returns PR_TRUE if 'attr' is present in the search requested attributes @@ -773,44 +774,39 @@ replica_config_search (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter { multimaster_mtnode_extension *mtnode_ext; int changeCount = 0; - PRBool reapActive = PR_FALSE; + PRBool reapActive = PR_FALSE; char val [64];
- /* add attribute that contains number of entries in the changelog for this replica */ + /* add attribute that contains number of entries in the changelog for this replica */
PR_Lock (s_configLock);
- mtnode_ext = _replica_config_get_mtnode_ext (e); - PR_ASSERT (mtnode_ext); - - if (mtnode_ext->replica) { - Replica *replica; - object_acquire (mtnode_ext->replica); - if (cl5GetState () == CL5_STATE_OPEN) { - changeCount = cl5GetOperationCount (mtnode_ext->replica); - } - replica = (Replica*)object_get_data (mtnode_ext->replica); - if (replica) { - reapActive = replica_get_tombstone_reap_active(replica); - } - - /* Check if the in memory ruv is requested */ - if (search_requested_attr(pb, type_ruvElement)) { - replica_config_search_ruv(pb, e, replica); - } - - /* Check if the last update time is requested */ - if (search_requested_attr(pb, type_ruvElementUpdatetime)) { - replica_config_search_last_modified(pb, e, replica); - } - - - object_release (mtnode_ext->replica); - } + mtnode_ext = _replica_config_get_mtnode_ext (e); + PR_ASSERT (mtnode_ext);
+ if (mtnode_ext->replica) { + Replica *replica; + object_acquire (mtnode_ext->replica); + if (cl5GetState () == CL5_STATE_OPEN) { + changeCount = cl5GetOperationCount (mtnode_ext->replica); + } + replica = (Replica*)object_get_data (mtnode_ext->replica); + if (replica) { + reapActive = replica_get_tombstone_reap_active(replica); + } + /* Check if the in memory ruv is requested */ + if (search_requested_attr(pb, type_ruvElement)) { + replica_config_search_ruv(pb, e, replica); + } + /* Check if the last update time is requested */ + if (search_requested_attr(pb, type_ruvElementUpdatetime)) { + replica_config_search_last_modified(pb, e, replica); + } + object_release (mtnode_ext->replica); + } sprintf (val, "%d", changeCount); slapi_entry_add_string (e, type_replicaChangeCount, val); - slapi_entry_attr_set_int(e, "nsds5replicaReapActive", (int)reapActive); + slapi_entry_attr_set_int(e, "nsds5replicaReapActive", (int)reapActive);
PR_Unlock (s_configLock);
diff --git a/ldap/servers/plugins/replication/repl5_ruv.c b/ldap/servers/plugins/replication/repl5_ruv.c index b68223b..d6a7448 100644 --- a/ldap/servers/plugins/replication/repl5_ruv.c +++ b/ldap/servers/plugins/replication/repl5_ruv.c @@ -1285,6 +1285,7 @@ ruv_to_smod(const RUV *ruv, Slapi_Mod *smod) int cookie; #define B_SIZ 1024 char buf[B_SIZ]; + slapi_rwlock_rdlock (ruv->lock); slapi_mod_init (smod, dl_get_count (ruv->elements) + 1); slapi_mod_set_type (smod, type_ruvElement); diff --git a/ldap/servers/plugins/replication/repl_globals.c b/ldap/servers/plugins/replication/repl_globals.c index c0b8712..ee09e49 100644 --- a/ldap/servers/plugins/replication/repl_globals.c +++ b/ldap/servers/plugins/replication/repl_globals.c @@ -105,6 +105,7 @@ const char *attr_flags = "nsds5Flags"; const char *attr_replicaName = "nsds5ReplicaName"; const char *attr_replicaReferral = "nsds5ReplicaReferral"; const char *type_ruvElement = "nsds50ruv"; +const char *type_agmtMaxCSN = "nsds5AgmtMaxCSN"; const char *type_replicaPurgeDelay = "nsds5ReplicaPurgeDelay"; const char *type_replicaChangeCount = "nsds5ReplicaChangeCount"; const char *type_replicaTombstonePurgeInterval = "nsds5ReplicaTombstonePurgeInterval"; diff --git a/ldap/servers/slapd/entry.c b/ldap/servers/slapd/entry.c index 59f2340..644026f 100644 --- a/ldap/servers/slapd/entry.c +++ b/ldap/servers/slapd/entry.c @@ -2871,13 +2871,15 @@ slapi_entry_attr_get_charptr( const Slapi_Entry* e, const char *type) { char *p= NULL; Slapi_Attr* attr = NULL; + slapi_entry_attr_find(e, type, &attr); if(attr!=NULL) { Slapi_Value *v; - const struct berval *bvp; + const struct berval *bvp; + slapi_valueset_first_value( &attr->a_present_values, &v); - bvp = slapi_value_get_berval(v); + bvp = slapi_value_get_berval(v); p= slapi_ch_malloc(bvp->bv_len + 1); memcpy(p, bvp->bv_val, bvp->bv_len); p[bvp->bv_len]= '\0'; diff --git a/ldap/servers/slapd/rdn.c b/ldap/servers/slapd/rdn.c index fe2fae0..32d783b 100644 --- a/ldap/servers/slapd/rdn.c +++ b/ldap/servers/slapd/rdn.c @@ -1096,3 +1096,13 @@ slapi_rdn_get_value(const char *rdn) return p; }
+char * +slapi_rdn_get_value_by_ref(const char *rdn) +{ + char *p = PL_strchr(rdn, '='); + if (p) { + p++; + while (isspace(*p)) p++; + } + return p; +} diff --git a/ldap/servers/slapd/slapi-plugin.h b/ldap/servers/slapd/slapi-plugin.h index 1ed92fa..497ce8f 100644 --- a/ldap/servers/slapd/slapi-plugin.h +++ b/ldap/servers/slapd/slapi-plugin.h @@ -3510,13 +3510,21 @@ int slapi_rdn_partial_dup(Slapi_RDN *from, Slapi_RDN **to, int idx); size_t slapi_rdn_get_size(Slapi_RDN *srdn);
/** - * Return the value of the RDN + * Return a copy of the value of the RDN * * \param rdn A pointer to rdn to get the value. * \return The value of the given RDN. */ char * slapi_rdn_get_value(const char *rdn);
+/** + * Return a pointer to the value of the RDN + * + * \param rdn A pointer to rdn to get the value. + * \return The value of the given RDN. + */ +char * slapi_rdn_get_value_by_ref(const char *rdn); + /* * utility routines for dealing with DNs */