Skip to content

[#942] Warn once per interval that a change is being retried, not once per delivery #982

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vharseko wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[#942] Warn once per interval that a change is being retried, not once per delivery #982

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 96 additions & 12 deletions ...rver-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -338,6 +338,14 @@ && getBackend().getBackendID().equals(backend.getBackendID())) {
* in flight unreplayable, and one alert per change would be a storm.
*/
private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000;
/**
* How long the warning telling that a change is being asked for again is not logged
* again, for the reason the alert above is not sent again and for one more: a change
* which keeps failing is asked for again every {@link #MAX_REPLAY_RETRY_DELAY_IN_MS}
* at the slowest, for as long as its give-up budget lasts - and how long that is has
* been the administrator's to set since issue #901.
*/
private static final long REPLAY_RETRY_WARNING_INTERVAL_IN_MS = 60000;
/**
* What the ack of a delivery whose replay ran out of memory says did not apply the
* change.
Expand Down Expand Up @@ -489,14 +497,24 @@ && getBackend().getBackendID().equals(backend.getBackendID())) {
*/
private volatile long replayDrainTimeoutInMs = REPLAY_DRAIN_TIMEOUT_IN_MS;
/**
* Stands for "the alert about a change this replica gave up on was never sent". The
* time it is compared with only moves forward from an origin which is arbitrary, so
* zero is not far enough in the past to say it.
* Stands for "a replay failure was never reported yet", whether by the alert about a
* change this replica gave up on or by the warning about a change it asks for again.
* The time it is compared with only moves forward from an origin which is arbitrary,
* so zero is not far enough in the past to say it.
*/
private static final long UNREPLAYED_CHANGE_ALERT_NEVER_SENT = Long.MIN_VALUE / 2;
private static final long REPLAY_FAILURE_NEVER_REPORTED = Long.MIN_VALUE / 2;
/** When the alert about a change this replica gave up on was last sent. */
private final AtomicLong lastUnreplayedChangeAlertTime =
new AtomicLong(UNREPLAYED_CHANGE_ALERT_NEVER_SENT);
new AtomicLong(REPLAY_FAILURE_NEVER_REPORTED);
/** When the warning about a change this replica asks for again was last logged. */
private final AtomicLong lastReplayRetryWarningTime =
new AtomicLong(REPLAY_FAILURE_NEVER_REPORTED);
/**
* How many failed deliveries were not warned about since the last warning was logged.
* They are counted rather than dropped: the line which is logged next says how many
* deliveries it stands for.
*/
private final AtomicInteger foldedReplayRetryWarnings = new AtomicInteger();
/**
* The result codes conflict resolution knows how to solve. The result code the server
* puts on an internal error is configurable and is not validated as a result code, so
Expand Down Expand Up @@ -2269,7 +2287,7 @@ void synchronize(PostOperationOperation op)
logger.error(ERR_OPERATION_NOT_FOUND_IN_PENDING, op, curCSN);
return;
}
resetSessionRestartBackoff();
resetReplayFailureTracking();
}
else
{
Expand Down Expand Up @@ -3387,25 +3405,33 @@ private static long monotonicNowInMs()
private void recordChangeResolved(CSN csn)
{
updateError(csn);
resetSessionRestartBackoff();
resetReplayFailureTracking();
}

/**
* Has the change which fails next start the backoff between the session restarts over,
* if this replica is not failing any change anymore.
* and the warning about it count the deliveries it stands for from zero, if this
* replica is not failing any change anymore.
* <p>
* A change made it and nothing is failing anymore, so the backend is serving again and
* the session is not being restarted in a row. While something is still failing, a
* change which was replayed says nothing of the kind - a change which can never be
* applied here fails alone, among changes which replay perfectly well, and letting
* those reset the wait would have this domain tear its session down every second for as
* long as that one change takes to be given up on.
* <p>
* The deliveries which were folded into no warning go with the backoff rather than into
* the next warning: a line logged when this domain fails again - a day later, over
* another change - would have them read as deliveries of that failure. How long the
* warning is not logged again is deliberately left alone, so that a backend which fails
* and recovers in turn is not one warning per failure again.
*/
private void resetSessionRestartBackoff()
private void resetReplayFailureTracking()
{
if (!remotePendingChanges.hasFailingChanges())
{
consecutiveSessionRestarts.set(0);
foldedReplayRetryWarnings.set(0);
}
}

Expand Down Expand Up @@ -3468,7 +3494,62 @@ private void sendUnreplayedChangeAlert(LocalizableMessage cause)
@VisibleForTesting
public void resetUnreplayedChangeAlertThrottle()
{
lastUnreplayedChangeAlertTime.set(UNREPLAYED_CHANGE_ALERT_NEVER_SENT);
lastUnreplayedChangeAlertTime.set(REPLAY_FAILURE_NEVER_REPORTED);
}

/**
* Warns that this replica could not replay a change and is asking for it again.
* <p>
* The warning is not logged again for {@link #REPLAY_RETRY_WARNING_INTERVAL_IN_MS},
* for the reason the alert above is not sent again and for one more: a change which
* keeps failing is delivered again every {@link #MAX_REPLAY_RETRY_DELAY_IN_MS} at the
* slowest, so one line per delivery is the same warning every ten seconds, for as long
* as the give-up budget of the change lasts - a budget the administrator sets, and
* which can be unlimited (issue #942).
* <p>
* The deliveries which are not warned about are counted rather than dropped, so the
* line which is logged says how many of them it stands for, and they are traced for
* whoever turns the replication debug logging on.
*
* @param csn the CSN of the change which could not be replayed
* @param failure how long, and over how many deliveries, its replay has been failing
*/
private void logReplayRetryWarning(CSN csn, RemotePendingChanges.ReplayFailure failure)
{
final long now = monotonicNowInMs();
final long lastLogged = lastReplayRetryWarningTime.get();
if (now - lastLogged >= REPLAY_RETRY_WARNING_INTERVAL_IN_MS
&& lastReplayRetryWarningTime.compareAndSet(lastLogged, now))
{
logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts(),
failure.getFailingForMs(), foldedReplayRetryWarnings.getAndSet(0));
}
else
{
/*
* A failure which loses the race against the thread which is logging right now is
* counted for the next line rather than for the one being written: what the count
* says is how many deliveries went unlogged, and carrying one of them over to the
* next interval says nothing else.
*/
foldedReplayRetryWarnings.incrementAndGet();
logger.trace("Could not replay change %s in domain %s: delivery %d, failing for %d ms",
csn, getBaseDN(), failure.getAttempts(), failure.getFailingForMs());
}
}

/**
* Lets the next change whose replay fails be warned about straight away.
* <p>
* Only there for the tests which check the warning: they must not be at the mercy of
* the warning another test logged less than
* {@link #REPLAY_RETRY_WARNING_INTERVAL_IN_MS} ago.
*/
@VisibleForTesting
public void resetReplayRetryWarningThrottle()
{
lastReplayRetryWarningTime.set(REPLAY_FAILURE_NEVER_REPORTED);
foldedReplayRetryWarnings.set(0);
}

/**
Expand Down Expand Up @@ -3584,9 +3665,12 @@ private boolean recoverFromReplayFailure(
* Not on the road out of a JVM which has run out of memory: building this line asks
* it for the memory it has just refused, and the ack of the delivery already says
* that the change was not applied. The constant that ack carries exists for the same
* reason.
* reason. The throttle is left alone as well - the trace line a folded delivery is
* written to asks for that memory too, and this delivery is not one which goes
* unlogged: the error ends the replay thread, and the uncaught exception handler of
* DirectoryThread writes the line and raises the alert for it.
*/
logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts());
logReplayRetryWarning(csn, failure);
}
/*
* This change is not owned by anyone anymore, so the session has to be restarted for
Expand Down
5 changes: 3 additions & 2 deletions opendj-server-legacy/src/messages/org/opends/messages/replication.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -617,8 +617,9 @@ ERR_COULD_NOT_BIND_CHANGELOG_PORT_FREE_305=Nothing holds %s anymore : the port w
NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED_306=Cannot start total update \
in domain "%s" from this directory server DS(%d): rejecting the request from the remote directory server DS(%d): %s
WARN_REPLAY_RETRYING_CHANGE_307=Could not replay change %s in domain "%s" (delivery %d, each \
attempted several times in place). The change has not been recorded as replayed: restarting the \
session to the replication server so that it is sent again
attempted several times in place, failing for %d ms). The change has not been recorded as \
replayed: restarting the session to the replication server so that it is sent again. %d further \
deliveries failed in this domain without being logged since the previous warning
ERR_REPLAY_SKIPPING_CHANGE_308=Could not replay change %s in domain "%s": its replay has been \
failing for %d ms over %d deliveries, each attempted several times in place. The change is being \
skipped: this replica now diverges from the rest of the topology and must be reinitialized. \
Expand Down
157 changes: 157 additions & 0 deletions opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import static org.forgerock.opendj.ldap.requests.Requests.*;
import static org.forgerock.opendj.ldap.schema.CoreSchema.*;
import static org.mockito.Mockito.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.opends.server.replication.plugin.LDAPReplicationDomain.*;
Expand All @@ -34,6 +35,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeoutException;
Expand Down Expand Up @@ -3615,6 +3617,161 @@ public CSN getCSN()
}
}

/**
* Test case for [Issue 942]: a change whose replay keeps failing is warned about once
* per interval rather than once per delivery.
* <p>
* The session is left down for ten seconds at the longest between two deliveries, so a
* warning per delivery is the same line every ten seconds for as long as the change is
* retried - and how long that is has been the administrator's to set since #901.
*/
@Test
public void aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval() throws Exception
{
testSetUp("aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval");
logger.error(LocalizableMessage.raw(
"Starting replication test : aChangeWhichKeepsFailingIsWarnedAboutOncePerInterval"));

final int serverId = 19;
ReplicationBroker broker =
openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
try
{
CSNGenerator gen = new CSNGenerator(serverId, 0);

Entry tmp = TestCaseUtils.addEntry(
"dn: uid=user.942," + baseDN,
"objectClass: top",
"objectClass: person",
"objectClass: organizationalPerson",
"objectClass: inetOrgPerson",
"uid: user.942",
"cn: Aaccf Amar",
"sn: Amar");
String uuid = getEntry(tmp.getName(), 1, true).parseAttribute("entryuuid").asString();

final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
/*
* The throttle is the domain's and the domain outlives the test methods: a change
* another test was retrying less than an interval ago would have the first warning
* of this one folded into its own.
*/
domain.resetReplayRetryWarningThrottle();
final CSN csn = gen.newCSN();
try
{
ShortCircuitPlugin.registerShortCircuit(
OperationType.DELETE, "PreParse", ResultCode.OTHER.intValue());
broker.publish(new DeleteMsg(tmp.getName(), csn, uuid));

/*
* A delivery burns IN_PLACE_REPLAY_ATTEMPTS short circuits before the session is
* restarted and the change is asked for again, so three times that many of them
* are three deliveries which failed - and three warnings, before this one was
* throttled. The give-up budget is minutes and the interval is a minute, so the
* change is still being retried by then and the deliveries all fall into one
* interval.
*/
TestTimer timer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(100, MILLISECONDS)
.toTimer();
timer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(ShortCircuitPlugin.getShortCircuitCount(OperationType.DELETE, "PreParse")
> 3 * IN_PLACE_REPLAY_ATTEMPTS,
"the change was not delivered again after its replay failed");
}
});
assertEquals(replayRetryWarnings(csn).size(), 1,
"a change which keeps failing must be warned about once per interval, not once per delivery");

/*
* Once the interval has passed the change is warned about again: a domain which
* never gives up - the budget can be unlimited - must not go silent over a change
* it is still asking for, and the line which comes says how many deliveries went
* unlogged in the meantime.
*/
TestTimer intervalTimer = new TestTimer.Builder()
.maxSleep(120, SECONDS)
.sleepTimes(500, MILLISECONDS)
.toTimer();
intervalTimer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertEquals(replayRetryWarnings(csn).size(), 2,
"a change which is still failing an interval later must be warned about again");
}
});
Assertions.assertThat(replayRetryWarnings(csn).get(1))
.as("the warning must say how many failed deliveries it stands for")
.containsPattern("[1-9]\\d* further deliveries failed");
}
finally
{
ShortCircuitPlugin.deregisterShortCircuit(OperationType.DELETE, "PreParse");
}

/*
* The backend serves again, so the delivery which comes next replays the change: a
* change left failing here would be the next test's, holding its ServerState back
* and its session restart backoff up.
*/
TestTimer replayTimer = new TestTimer.Builder()
.maxSleep(60, SECONDS)
.sleepTimes(200, MILLISECONDS)
.toTimer();
replayTimer.repeatUntilSuccess(new CallableVoid()
{
@Override
public void call() throws Exception
{
assertTrue(domain.getServerState().cover(csn),
"the change must be replayed once the backend serves again");
}
});
}
finally
{
broker.stop();
}
}

/**
* Returns the warnings this replica logged about the provided change being asked for
* again, oldest first.
* <p>
* The error log of the test server is written to a writer which keeps every record, so
* the warnings about one change are the records which carry the ordinal of the message
* and the CSN of the change.
* <p>
* The test server registers two error log publishers over that one writer, so it keeps
* every record twice: what is returned here is the records which differ. Two warnings
* about the same change never read the same - the delivery they report, how long the
* change has been failing and how many deliveries were folded into them all move on.
*
* @param csn the CSN of the change whose replay keeps failing
* @return the warnings which name it, in the order they were logged
*/
private static List<String> replayRetryWarnings(CSN csn)
{
final String messageId = "msgID=" + WARN_REPLAY_RETRYING_CHANGE.ordinal();
final Set<String> warnings = new LinkedHashSet<>();
for (String record : TestCaseUtils.ERROR_TEXT_WRITER.getMessages())
{
if (record.contains(messageId) && record.contains(csn.toString()))
{
warnings.add(record);
}
}
return new ArrayList<>(warnings);
}

/**
* Test case for [Issue 908]: a domain being disabled - for an LDIF import, a restore, or
* a backend being taken offline - must not save its ServerState while a replay thread is
Expand Down
Loading