StateMachineTest.java revision 9aa1482c49865483f798f91ef7d16f03cc3ec811
/*
* CDDL HEADER START
*
* The contents of this file are subject to the terms of the
* Common Development and Distribution License, Version 1.0 only
* (the "License"). You may not use this file except in compliance
* with the License.
*
* You can obtain a copy of the license at legal-notices/CDDLv1_0.txt
* See the License for the specific language governing permissions
* and limitations under the License.
*
* When distributing Covered Code, include this CDDL HEADER in each
* file and include the License file at legal-notices/CDDLv1_0.txt.
* If applicable, add the following below this CDDL HEADER, with the
* fields enclosed by brackets "[]" replaced with your own identifying
* information:
* Portions Copyright [yyyy] [name of copyright owner]
*
* CDDL HEADER END
*
*
* Copyright 2009-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2015 ForgeRock AS
*/
/**
* Some tests to go through the DS state machine and validate we get the
* expected status according to the actions we perform.
*/
@SuppressWarnings("javadoc")
public class StateMachineTest extends ReplicationTestCase
{
/** Server id definitions. */
private static DN EXAMPLE_DN_;
private static final int DS1_ID = 1;
private static final int DS2_ID = 2;
private static final int DS3_ID = 3;
private static final int RS1_ID = 41;
private int rs1Port = -1;
private LDAPReplicationDomain ds1;
private ReplicationBroker ds2;
private ReplicationBroker ds3;
private ReplicationServer rs1;
/** The tracer object for the debug logger. */
private final int initWindow = 100;
{
if (logger.isTraceEnabled())
{
}
}
{
{
}
}
{
{
}
}
private void initTest() throws IOException
{
rs1Port = -1;
}
{
{
}
// Clear any reference to a domain in synchro plugin
rs1Port = -1;
}
/** Waits until the provided ds is connected to the replication server. */
{
.toTimer();
{
{
return null;
}
});
}
private boolean isConnected(final int dsId)
{
switch (dsId)
{
case DS1_ID:
return ds1.isConnected();
case DS2_ID:
return ds2.isConnected();
case DS3_ID:
return ds3.isConnected();
default:
fail("Unknown ds server id.");
return false;
}
}
/** Creates a new ReplicationServer. */
int degradedStatusThreshold) throws Exception
{
return new ReplicationServer(conf);
}
/** Creates and starts a new ReplicationDomain configured for the replication server. */
@SuppressWarnings("unchecked")
{
if (provider instanceof ConfigurationChangeListener)
{
}
return replicationDomain;
}
/**
* Create and connect a replication broker to the replication server with the
* given state and generation id.
*/
{
return broker;
}
/**
* Make simple state machine test.
*
* NC = Not connected status
* N = Normal status
* D = Degraded status
* FU = Full update status
* BG = Bad generation id status
*
* The test path should be:
* ->NC->N->NC
* @throws Exception If a problem occurred
*/
public void testStateMachineBasic() throws Exception
{
initTest();
try
{
// DS1 start, no RS available: DS1 should be in not connected status
// RS1 starts , DS1 should connect to it and be in normal status
// RS1 stops, DS1 should go in not connected status
} finally
{
endTest();
}
}
/** Returns various init values for test testStateMachineStatusAnalyzer. */
{
}
/**
* Test the status analyzer system that allows to go from normal to degraded
* and vice versa, using the configured threshold value
*
* NC = Not connected status
* N = Normal status
* D = Degraded status
* FU = Full update status
* BG = Bad generation id status
*
* Expected path:
* ->NC->N->D->N->NC
* @throws Exception If a problem occurred
*/
{
initTest();
try
{
/* RS1 starts with specified threshold value */
/*
* DS2 starts and connects to RS1. No reader and low window value at the beginning
* so writer for DS2 in RS should enqueue changes after first changes sent to DS.
* (window value reached: a window msg needed by RS for following sending changes to DS)
*/
/* DS3 starts and connects to RS1 */
// Send first changes to reach window and block DS2 writer queue. Writer will take them
// from queue and block (no more changes removed from writer queue) after
// having sent them to TCP receive queue of DS2.
/*
* DS3 sends changes (less than threshold): DS2 should still be in normal status
* so no topo message should be sent (update topo message for telling status of DS2 changed)
*/
int nChangesSent = 0;
if (thresholdValue > 1)
{
}
/*
* DS3 sends changes to reach the threshold value,
* DS3 should receive an update topo message with status of DS2: degraded status
*/
// wait for a status MSG status analyzer to broker 3
/*
* DS3 sends 10 additional changes after threshold value,
* DS2 should still be degraded so no topo message received.
*/
/*
* DS2 replays every changes and should go back to normal status
* (create a reader to emulate replay of messages (messages read from queue))
*/
// wait for a status MSG status analyzer to broker 3
} finally
{
endTest();
}
}
private void waitForDegradedStatusOnBroker3() throws InterruptedException
{
{
{
break;
}
}
}
/**
* Go through the possible state machine transitions:
*
* NC = Not connected status
* N = Normal status
* D = Degraded status
* FU = Full update status
* BG = Bad generation id status
*
* The test path should be:
* ->NC->D->N->NC->N->D->NC->D->N->BG->NC->N->D->BG->FU->NC->N->D->FU->NC->BG->NC->N->FU->NC->N->NC
* @throws Exception If a problem occurred
*/
public void testStateMachineFull() throws Exception
{
initTest();
try
{
int DEGRADED_STATUS_THRESHOLD = 1;
// RS1 starts with 1 message as degraded status threshold value
// DS2 starts and connects to RS1
// DS2 starts sending a lot of changes
/*
* DS1 starts and connects to RS1, server state exchange should lead to start in degraded status
* as some changes should be in queued in the RS and the threshold value is 1 change in queue.
*/
/* DS2 stops sending changes: DS1 should replay pending changes and should enter the normal status */
// Sleep enough so that replay can be done and analyzer has time
// to see that the queue length is now under the threshold value.
/* RS1 stops to make DS1 go to not connected status (from normal status) */
/*
* DS2 restarts with up to date server state
* (this allows to have restarting RS1 not sending him some updates he already sent)
*/
/* RS1 restarts, DS1 should get back to normal status */
/* DS2 sends again a lot of changes to make DS1 degraded again */
/* RS1 stops to make DS1 go to not connected status (from degraded status) */
/*
* DS2 restarts with up to date server state
* (this allows to have restarting RS1 not sending him some updates he already sent)
*/
/*
* RS1 restarts, DS1 should reconnect in degraded status
* (from not connected this time, not from state machine entry)
*/
// It is too difficult to tune the right sleep so disabling this test:
// Sometimes the status analyzer may be fast and quickly change the status
// of DS1 to NORMAL_STATUS
//sleep(2000);
//sleepAssertStatusEquals(30, ds1, ServerStatus.DEGRADED_STATUS);
/* DS1 should come back in normal status after a while */
/*
* DS2 sends a reset gen id order with wrong gen id:
* DS1 should go into bad generation id status
*/
long BAD_GEN_ID = 999999L;
/*
* DS2 sends again a reset gen id order with right id: DS1 should be disconnected by RS
* then reconnect and enter again in normal status. This goes through not connected status
* but not possible to check as should reconnect immediately
*/
// broker to send changex
/* DS2 sends again a lot of changes to make DS1 degraded again */
/*
* DS2 sends reset gen id order with bad gen id: DS1 should go in bad gen id status
* (from degraded status this time)
*/
resetGenId(ds2, -1); // -1 to allow next step full update and flush RS db so that DS1 can reconnect after full update
/*
* DS2 engages full update (while DS1 in bad gen id status),
* DS1 should go in full update status
*/
/*
* DS2 terminates full update to DS1: DS1 should reconnect (goes through not connected status)
* and come back to normal status (RS genid was -1 so RS will adopt new gen id)
*/
bi.runFullUpdate();
/* DS2 sends changes to DS1: DS1 should go in degraded status */
/* DS2 engages full update (while DS1 in degraded status), DS1 should go in full update status */
/*
* DS2 terminates full update to DS1: DS1 should reconnect (goes through not connected status)
* and come back to bad gen id status (RS genid was another gen id (300 entries instead of 200)
*/
bi.runFullUpdate();
/*
* DS2 sends reset gen id with gen id same as DS1:
* DS1 will be disconnected by RS (not connected status) and come back to normal status
*/
/* DS2 engages full update (while DS1 in normal status), DS1 should go in full update status */
/*
* DS2 terminates full update to DS1: DS1 should reconnect (goes through not connected status)
* and come back to normal status (process full update with same data as before so RS already
* has right gen id: version with 300 entries)
*/
bi.runFullUpdate();
/* RS1 stops, DS1 should go to not connected status */
} finally
{
// Finalize test
endTest();
}
}
/**
* Set up the environment.
*
* @throws Exception
* If the environment could not be set up.
*/
{
super.setUp();
// Note: this test does not use the memory test backend as for having a DS
// going into degraded status, we need to send a lot of updates. This makes
// the memory test backend crash with OutOfMemoryError. So we prefer here
// a backend backed up with a file
}
/**
* Clean up the environment.
*
* @throws Exception If the environment could not be set up.
*/
public void classCleanUp() throws Exception
{
callParanoiaCheck = false;
super.classCleanUp();
}
/**
* Sends a reset genid message through the given replication broker, with the
* given new generation id.
*/
{
}
/**
* Utility class for making a full update through a broker. No separated thread
* Usage:
* BrokerInitializer bi = new BrokerInitializer(rb, sid, nEntries);
* bi.initFullUpdate(); // Initializes a full update session by sending InitializeTargetMsg
* bi.runFullUpdate(); // loops sending nEntries entries and finalizes the full update by sending the EntryDoneMsg
*/
private class BrokerInitializer
{
private ReplicationBroker rb;
private int serverId = -1;
private long userId;
/** Server id of server to initialize. */
private int destId = -1;
/** Number of entries to send to dest. */
private long nEntries = -1;
private boolean createReader;
/**
* If the BrokerInitializer is to be used for a lot of entries to send
* (which is often the case), the reader thread should be enabled to make
* the window subsystem work and allow the broker to send as much entries as
* he wants. If not enabled, the user is responsible to call the receive
* method of the broker himself.
*/
private BrokerReader reader;
/** Creates a broker initializer. Also creates a reader according to request */
boolean createReader)
{
this.createReader = createReader;
}
/** Initializes a full update session by sending InitializeTargetMsg. */
{
// Also create reader ?
if (createReader)
{
}
// Send init msg to warn dest server it is going do be initialized
// Send top entry for the domain
+ "objectClass: top\n"
+ "objectClass: domain\n"
+ "dc: example\n"
+ "entryUUID: 11111111-1111-1111-1111-111111111111\n\n";
}
private EntryMsg createNextEntryMsg()
{
"objectClass: person\n" + "objectClass: organizationalPerson\n" +
"objectClass: inetOrgPerson\n" +
"homePhone: 951-245-7634\n" +
"description: This is the description for Aaccf Amar.\n" + "st: NC\n" +
"mobile: 027-085-0537\n" +
"postalAddress: Aaccf Amar$17984 Thirteenth Street" +
"$Rockford, NC 85762\n" + "mail: user.1@example.com\n" +
"cn: Aaccf Amar\n" + "l: Rockford\n" + "pager: 508-763-4246\n" +
"street: 17984 Thirteenth Street\n" + "telephoneNumber: 216-564-6748\n" +
"employeeNumber: 1\n" + "sn: Amar\n" + "givenName: Aaccf\n" +
"postalCode: 85762\n" + "userPassword: password\n" + "initials: AA\n" +
// -> WARNING: EntryMsg PDUs are concatenated before calling import on LDIF
// file so need \n\n to separate LDIF entries to conform to LDIF file format
// Create an entry message
(int) userId);
}
/**
* Loops sending entries for full update (EntryMsg messages). When
* terminates, sends the EntryDoneMsg to finalize full update. Number of
* sent entries is determined at initFullUpdate call time.
*/
public void runFullUpdate()
{
for(long i = 0 ; i<nEntries ; i++) {
}
if (createReader)
{
}
}
}
/** Thread for sending a lot of changes through a broker. */
private class BrokerWriter extends Thread
{
private ReplicationBroker rb;
private int serverId = -1;
private long userId;
/** The writer starts suspended. */
/**
* Tells a sending session is finished. A session is sending messages
* between the follow and the pause calls, or the time a followAndPause
* method runs.
*/
private boolean careAboutAmountOfChanges;
/** Number of sent changes. */
private int nChangesSent;
private int nChangesSentLimit;
private CSNGenerator gen;
/**
* If the BrokerWriter is to be used for a lot of changes to send (which is
* often the case), the reader thread should be enabled to make the window
* subsystem work and allow the broker to send as much changes as he wants.
* If not enabled, the user is responsible to call the receive method of
* the broker himself.
*/
private BrokerReader reader;
/** Creates a broker writer. Also creates a reader according to request */
boolean createReader)
{
super("BrokerWriter for broker " + serverId);
// Create a csn generator to generate new csns
// when we need to send changes
// Start thread (is paused by default so will have to call follow anyway)
start();
// Also create reader ?
if (createReader)
{
}
}
/**
* Loops sending changes: add operations creating users with different ids
* This starts paused and has to be resumed calling a follow method.
*/
public void run()
{
boolean dbg1Written = false, dbg2Written;
// No stop msg when entering the loop (thread starts with paused writer)
dbg2Written = true;
{
long startSessionTime = -1;
boolean startedNewSession = false;
// When not in pause, loop sending changes to RS
{
startedNewSession = true;
if (!dbg1Written)
{
" writer starting sending changes session at: " + startSessionTime);
dbg1Written = true;
dbg2Written = false;
}
// End session if amount of changes sent has been requested
{
nChangesSent++;
if (nChangesSent == nChangesSentLimit)
{
// Requested number of changes to send sent, end session
nChangesSent + " changes limit");
break;
}
}
}
if (!dbg2Written)
{
" writer stopping sending changes session at: " + endSessionTime +
dbg1Written = false;
dbg2Written = true;
}
// Mark session is finished
if (startedNewSession)
{
sessionDone.set(true);
}
try
{
// Writer in pause, sleep a while to let other threads work
synchronized(sleeper)
{
}
} catch (InterruptedException ex)
{
/* Don't care */
}
}
}
/** Stops the writer thread. */
public void shutdown()
{
synchronized (sleeper)
{
}
try
{
join();
} catch (InterruptedException ex)
{
/* Don't care */
}
// Stop reader if any
}
/** Suspends the writer thread. */
public void pause()
{
if (isPaused())
{
return; // Already suspended
}
// Wait for all messages sent
while (!sessionDone.get())
{
}
}
/** Test if the writer is suspended. */
public boolean isPaused()
{
return sessionDone.get();
}
/** Resumes the writer thread until it is paused. */
public void follow()
{
sessionDone.set(false);
}
/**
* Resumes the writer and suspends it after a given amount of changes has been
* sent. If the writer was working it will be paused anyway after the given
* amount of changes, starting from the current call time.
* -> blocking call
*/
public void followAndPause(int nChanges)
{
pause(); // If however we were already working
// Initialize counter system variables
nChangesSent = 0;
careAboutAmountOfChanges = true;
// Start session
sessionDone.set(false);
// Wait for all messages sent
while (!sessionDone.get())
{
}
careAboutAmountOfChanges = false;
}
private AddMsg createNextAddMsg()
{
"objectClass: person\n" + "objectClass: organizationalPerson\n" +
"objectClass: inetOrgPerson\n" +
"homePhone: 951-245-7634\n" +
"description: This is the description for Aaccf Amar.\n" + "st: NC\n" +
"mobile: 027-085-0537\n" +
"postalAddress: Aaccf Amar$17984 Thirteenth Street" +
"$Rockford, NC 85762\n" + "mail: user.1@example.com\n" +
"cn: Aaccf Amar\n" + "l: Rockford\n" + "pager: 508-763-4246\n" +
"street: 17984 Thirteenth Street\n" + "telephoneNumber: 216-564-6748\n" +
"employeeNumber: 1\n" + "sn: Amar\n" + "givenName: Aaccf\n" +
"postalCode: 85762\n" + "userPassword: password\n" + "initials: AA\n" +
try
{
} catch (Exception e)
{
throw new RuntimeException(e);
}
// Create an update message to add an entry.
null,
}
}
/**
* This simple reader just throws away the received
* messages. It is used on a breaker we want to be able to send or read from some message
* with (changes, entries (full update)...). Calling the receive method of the
* broker allows to unblock the window mechanism and to send the desired messages.
* Calling the updateWindowAfterReplay method allows to send when necessary the
* window message to the RS to allow him send other messages he may want to send us.
*/
private class BrokerReader extends Thread
{
private ReplicationBroker rb;
private int serverId = -1;
private boolean shutdown;
private ReplicationMsg lastMsg;
{
super("BrokerReader for broker " + serverId);
start();
}
/** Loop reading and throwing update messages. */
public void run()
{
while (!shutdown)
{
try
{
{
}
} catch (SocketTimeoutException ex)
{
if (shutdown)
{
return;
}
}
}
}
/** Returns last received message from reader When read, last value is cleared. */
public ReplicationMsg getLastMsg()
{
return toReturn;
}
/** Stops reader thread. */
public void shutdown()
{
shutdown = true;
try
{
join();
} catch (InterruptedException ex)
{
/* Don't care */
}
}
}
/**
* Waits until the domain status reaches the expected status.
* @param domain The domain whose status we want to test
* @param expectedStatus The expected domain status
*/
private void waitUntilStatusEquals(final LDAPReplicationDomain domain, final ServerStatus expectedStatus) throws Exception
{
.toTimer();
{
{
return null;
}
});
}
}