Add configuration to select which ranges we warm up. Default (null) is all ranges
Change-Id: I0f241563959cc9fb673ab6d8a87c45e8399086c2
Issue-ID: MUSIC-287
Signed-off-by: Tschaen, Brendan <ctschaen@att.com>
<version>1.0.0</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>2.23.4</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>com.github.jsqlparser</groupId>
<artifactId>jsqlparser</artifactId>
public static final String TX_DAEMON_SLEEPTIME_S = "txdaemonsleeps";
/** Default txDigest Daemon sleep time */
public static final String TX_DAEMON_SLEEPTIME_S_DEFAULT = "10";
+ /** The property name to use to provide a timeout to mdbc (ownership) */
+ public static final String KEY_OWNERSHIP_TIMEOUT = "mdbc_timeout";
+ /** The default property value to use for the MDBC timeout */
+ public static final long DEFAULT_OWNERSHIP_TIMEOUT = 5*60*60*1000;//default of 5 hours
+ /** The property name to provide comma separated list of ranges to warmup */
+ public static final String KEY_WARMUPRANGES = "warmupranges";
}
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
String cassandraUrl;
private Properties info;
- /** The property name to use to provide a timeout to mdbc (ownership) */
- public static final String KEY_TIMEOUT = "mdbc_timeout";
- /** The default property value to use for the MDBC timeout */
- public static final long DEFAULT_TIMEOUT = 5*60*60*1000;//default of 5 hours
-
/** Identifier for this server instance */
private String mdbcServerName;
private Map<String,DatabasePartition> connectionRanges;//Each connection owns its own database partition
private final Lock eventualLock = new ReentrantLock();
private List<Range> eventualRanges;
+ /** lock for warmupRanges */
private final Lock warmupLock = new ReentrantLock();
- private List<Range> warmupRanges;
+ /** a set of ranges that should be periodically updated with latest information, if null all tables should be warmed up */
+ private Set<Range> rangesToWarmup;
+ /** map of transactions that have already been applied/updated in this sites SQL db */
private Map<Range, Pair<MriReference, Integer>> alreadyApplied;
private OwnershipAndCheckpoint ownAndCheck;
+ /**
+ * For testing purposes only
+ */
+ @Deprecated
+ public StateManager() {
+ }
+
public StateManager(String sqlDBUrl, Properties info, String mdbcServerName, String sqlDBName) throws MDBCServiceException {
this.sqlDBName = sqlDBName;
this.sqlDBUrl = sqlDBUrl;
initMusic();
initSqlDatabase();
- String t = info.getProperty(KEY_TIMEOUT);
- long timeout = (t == null) ? DEFAULT_TIMEOUT : Integer.parseInt(t);
+ String t = info.getProperty(Configuration.KEY_OWNERSHIP_TIMEOUT);
+ long timeoutMs = (t == null) ? Configuration.DEFAULT_OWNERSHIP_TIMEOUT : Integer.parseInt(t);
alreadyApplied = new ConcurrentHashMap<>();
- ownAndCheck = new OwnershipAndCheckpoint(alreadyApplied, timeout);
+ ownAndCheck = new OwnershipAndCheckpoint(alreadyApplied, timeoutMs);
+ rangesToWarmup = initWarmupRanges();
+ logger.info("Warmup ranges for this site is " + rangesToWarmup);
+
MusicTxDigest txDaemon = new MusicTxDigest(this);
txDaemon.startBackgroundDaemon(Integer.parseInt(
info.getProperty(Configuration.TX_DAEMON_SLEEPTIME_S, Configuration.TX_DAEMON_SLEEPTIME_S_DEFAULT)));
throw new MDBCServiceException(e.getMessage(), e);
}
}
+
+ /**
+ * Get list of ranges to warmup from configuration file
+ * if no configuration is provided, will return null
+ * @return
+ */
+ private Set<Range> initWarmupRanges() {
+ String warmupString = info.getProperty(Configuration.KEY_WARMUPRANGES);
+ if (warmupString==null) {
+ return null;
+ }
+ Set<Range> warmupRanges = new HashSet<>();
+ String[] ranges = warmupString.split(",");
+ for (String range: ranges) {
+ warmupRanges.add(new Range(range.trim()));
+ }
+ return warmupRanges;
+ }
public MusicInterface getMusicInterface() {
return this.musicInterface;
return new ArrayList<>(connectionRanges.values());
}
- public List<Range> getWarmupRanges(){
+ /**
+ * Get a list of ranges that are to be periodically warmed up
+ *
+ * If no list is specified, all ranges except eventual consistency ranges are returned
+ * @return
+ */
+ public Set<Range> getRangesToWarmup() {
warmupLock.lock();
- List<Range> returnArray;
+ Set<Range> returnSet;
try {
- if(warmupRanges!=null) {
- returnArray = new ArrayList<>(warmupRanges);
+ if(rangesToWarmup!=null) {
+ returnSet = rangesToWarmup;
}
- else{
- returnArray = null;
+ else {
+ returnSet = getAllRanges();
+ for (Range eventualRange: eventualRanges) {
+ returnSet.remove(eventualRange);
+ }
}
}
finally{
warmupLock.unlock();
}
- return returnArray;
+ return returnSet;
}
+ /**
+ * Get a set of all ranges seen in the sql db
+ * @return
+ */
+ private Set<Range> getAllRanges() {
+ DBInterface dbi = ((MdbcConnection) getConnection("daemon")).getDBInterface();
+ return dbi.getSQLRangeSet();
+ }
+
+ /**
+ * Get a list of ranges that are eventually consistent
+ * @return
+ */
public List<Range> getEventualRanges() {
eventualLock.lock();
List<Range> returnArray;
this.mdbcServerName = mdbcServerName;
}
+ /**
+ * Close connection and relinquish any locks held for that connection
+ * @param connectionId
+ */
public void closeConnection(String connectionId){
//\TODO check if there is a race condition
if(mdbcConnections.containsKey(connectionId)) {
}
- public void setWarmupRanges(List<Range> warmupRanges) {
+ public void setWarmupRanges(Set<Range> warmupRanges) {
warmupLock.lock();
try {
- this.warmupRanges = warmupRanges;
+ this.rangesToWarmup = warmupRanges;
}
finally{
warmupLock.unlock();
System.exit(1);\r
}\r
Connection connection;\r
- try {\r
- connection = DriverManager.getConnection("jdbc:avatica:remote:url=http://localhost:30000/test;serialization=protobuf");\r
- } catch (SQLException e) {\r
- e.printStackTrace();\r
- return;\r
+ try { \r
+ String metricURL = "http://localhost:300000/test"; \r
+ if(args[0] != null) { \r
+ metricURL = args[0]; \r
+ } \r
+ connection = DriverManager.getConnection("jdbc:avatica:remote:url=" + metricURL+ ";serialization=protobuf"); \r
+ } catch (SQLException e) { \r
+ e.printStackTrace(); \r
+ return; \r
}\r
\r
try {\r
}\r
\r
final String insertSQL = "INSERT INTO Persons VALUES (1, 'Martinez', 'Juan', 'KACB', 'ATLANTA');";\r
- final String insertSQL1 = "DELETE FROM Persons WHERE PersonID=1;";\r
+ final String insertSQL1 = "DELETE FROM Persons WHERE PersonID=2;";\r
final String insertSQL2 = "INSERT INTO Persons VALUES (2, 'Smith', 'JOHN', 'GNOC', 'BEDMINSTER');";\r
final String insertSQL3 = "UPDATE Persons SET FirstName='JOSH' WHERE LastName='Smith';";\r
final String insertSQL4 = "UPDATE Persons SET FirstName='JOHN' WHERE LastName='Smith';";\r
}\r
\r
try {\r
- execute = insertStmt.execute(insertSQL);\r
+ //execute = insertStmt.execute(insertSQL);\r
execute = insertStmt.execute(insertSQL1);\r
- execute = insertStmt.execute(insertSQL2);\r
- execute = insertStmt.execute(insertSQL3);\r
- execute = insertStmt.execute(insertSQL4);\r
+ //execute = insertStmt.execute(insertSQL2);\r
+ //execute = insertStmt.execute(insertSQL3);\r
+ //execute = insertStmt.execute(insertSQL4);\r
\r
} catch (SQLException e) {\r
e.printStackTrace();\r
* Get a set of the table names in the database. The table names should be returned in UPPER CASE.
* @return the set
*/
+ @Deprecated
Set<String> getSQLTableSet();
/**
+ * Get a set of the ranges in the database
+ * @return the set
+ */
+ Set<Range> getSQLRangeSet();
+ /**
* Return the name of the database that the driver is connected to
* @return
*/
import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
public String getDatabaseName() {
return this.dbName;
}
- /**
- * Get a set of the table names in the database.
- * @return the set
- */
+
@Override
public Set<String> getSQLTableSet() {
Set<String> set = new TreeSet<String>();
logger.debug(EELFLoggerDelegate.applicationLogger,"getSQLTableSet returning: "+ set);
return set;
}
+
+ @Override
+ public Set<Range> getSQLRangeSet() {
+ Set<String> set = new TreeSet<String>();
+ String sql = "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_TYPE='BASE TABLE'";
+ try {
+ Statement stmt = jdbcConn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ while (rs.next()) {
+ String s = rs.getString("TABLE_NAME");
+ set.add(s);
+ }
+ stmt.close();
+ } catch (SQLException e) {
+ logger.error(EELFLoggerDelegate.errorLogger,"getSQLTableSet: "+e);
+ }
+ logger.debug(EELFLoggerDelegate.applicationLogger,"getSQLTableSet returning: "+ set);
+ Set<Range> rangeSet = new HashSet<>();
+ for (String table: set) {
+ rangeSet.add(new Range(table));
+ }
+ return rangeSet;
+ }
+
/*
mysql> describe tables;
+-----------------+---------------------+------+-----+---------+-------+
throw new MDBCServiceException("Error applying tx digest in local SQL",e);
}
}
-
- public void warmup(MusicInterface mi, DBInterface di, List<Range> ranges) throws MDBCServiceException {
- if(ranges.isEmpty()){
+
+ /**
+ * Replay the updates for the partitions containing ranges to the local database
+ * @param mi
+ * @param di
+ * @param rangesToWarmup
+ * @throws MDBCServiceException
+ */
+ public void warmup(MusicInterface mi, DBInterface di, List<Range> rangesToWarmup) throws MDBCServiceException {
+ if(rangesToWarmup.isEmpty()){
return;
}
boolean ready = false;
change.set(true);
- Set<Range> rangeSet = new HashSet<Range>(ranges);
+ Set<Range> rangeSet = new HashSet<Range>(rangesToWarmup);
Dag dag = new Dag(false);
while(!ready){
if(change.get()){
change.set(false);
- final List<MusicRangeInformationRow> rows = extractRowsForRange(mi, ranges,false);
- dag = Dag.getDag(rows,ranges);
+ final List<MusicRangeInformationRow> rows = extractRowsForRange(mi, rangesToWarmup,false);
+ dag = Dag.getDag(rows,rangesToWarmup);
}
else if(!dag.applied()){
- DagNode node = dag.nextToApply(ranges);
+ DagNode node = dag.nextToApply(rangesToWarmup);
if(node!=null) {
Pair<MusicTxDigestId, List<Range>> pair = node.nextNotAppliedTransaction(rangeSet);
while (pair != null) {
break;
} else {
final StagingTable txDigest = mi.getTxDigest(pair.getKey());
- applyTxDigest(ranges,di, txDigest);
+ applyTxDigest(rangesToWarmup,di, txDigest);
for (Range r : pair.getValue()) {
MusicRangeInformationRow row = node.getRow();
alreadyApplied.put(r, Pair.of(new MriReference(row.getPartitionIndex()), pair.getKey().index));
continue;
}
//2) for each partition I don't own
- final List<Range> warmuplist = stateManager.getWarmupRanges();
- if(warmuplist!=null) {
- final Set<Range> warmupRanges = new HashSet(warmuplist);
- final List<DatabasePartition> currentPartitions = stateManager.getPartitions();
- List<Range> missingRanges = new ArrayList<>();
- if (currentPartitions.size() != 0) {
- for (DatabasePartition part : currentPartitions) {
- List<Range> partitionRanges = part.getSnapshot();
- warmupRanges.removeAll(partitionRanges);
- }
- try {
- stateManager.getOwnAndCheck().warmup(mi, dbi, new ArrayList<>(warmupRanges));
- } catch (MDBCServiceException e) {
- logger.error("Unable to update for partition : " + warmupRanges + ". " + e.getMessage());
- continue;
- }
+ final Set<Range> warmupRanges = stateManager.getRangesToWarmup();
+ final List<DatabasePartition> currentPartitions = stateManager.getPartitions();
+ if (currentPartitions.size() != 0) {
+ for (DatabasePartition part : currentPartitions) {
+ List<Range> partitionRanges = part.getSnapshot();
+ warmupRanges.removeAll(partitionRanges);
+ }
+ try {
+ stateManager.getOwnAndCheck().warmup(mi, dbi, new ArrayList<>(warmupRanges));
+ } catch (MDBCServiceException e) {
+ logger.error("Unable to update for partition : " + warmupRanges + ". " + e.getMessage());
+ continue;
}
}
cassandra
cassandra.password =\
cassandra
-zookeeper.host =\
- localhost
+music_namespace =\
+ mdbc_namespace
\ No newline at end of file
--- /dev/null
+/*
+ * ============LICENSE_START====================================================
+ * org.onap.music.mdbc
+ * =============================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * =============================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END======================================================
+ */
+package org.onap.music.mdbc;
+
+import static org.junit.Assert.*;
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.internal.util.reflection.FieldSetter;
+import org.onap.music.exceptions.MDBCServiceException;
+import org.onap.music.mdbc.mixins.DBInterface;
+import org.onap.music.mdbc.tables.TxCommitProgress;
+
+public class StateManagerTest {
+
+ StateManager stateManager;
+
+ @BeforeClass
+ public static void beforeClass() {
+ System.out.println("StateManagerTest");
+ }
+
+ @Before
+ public void before() throws MDBCServiceException {
+ //shouldn't use separate constructor, but this will do for now
+ stateManager = new StateManager();
+ }
+
+ @Test
+ public void testGetEventualRanges() throws NoSuchFieldException, SecurityException {
+ List<Range> evList = new ArrayList<>();
+ evList.add(new Range("eventualRange"));
+ FieldSetter.setField(stateManager, stateManager.getClass().getDeclaredField("eventualRanges"), evList);
+ assertEquals(evList, stateManager.getEventualRanges());
+ }
+
+ @Test
+ public void testSetEventualRanges() {
+ List<Range> evList = new ArrayList<>();
+ evList.add(new Range("eventualRange"));
+ stateManager.setEventualRanges(evList);
+ assertEquals(evList, stateManager.getEventualRanges());
+ }
+
+ @Test
+ public void testSetMdbcServerName() {
+ String serverName = "serverName";
+ stateManager.setMdbcServerName(serverName);
+ assertEquals(serverName, stateManager.getMdbcServerName());
+ }
+
+ @Test
+ public void testGetConnection() throws Exception {
+ System.out.println("Testing getting a connection");
+
+ Connection connMock = Mockito.mock(Connection.class);
+ String connName = "connectionName";
+ Map<String, Connection> connMap = new HashMap<>();
+ connMap.put(connName, connMock);
+ FieldSetter.setField(stateManager, stateManager.getClass().getDeclaredField("mdbcConnections"),
+ connMap);
+
+ TxCommitProgress txInfoMock = Mockito.mock(TxCommitProgress.class);
+ FieldSetter.setField(stateManager, stateManager.getClass().getDeclaredField("transactionInfo"),
+ txInfoMock);
+
+
+ assertEquals(connMock, stateManager.getConnection(connName));
+ }
+
+ @Test
+ public void testGetRangesToWarmup() throws Exception {
+ System.out.println("Testing warmup ranges where no ranges are defined");
+
+ //getConnection
+ MdbcConnection connMock = Mockito.mock(MdbcConnection.class);
+ String connName = "daemon";
+ Map<String, Connection> connMap = new HashMap<>();
+ connMap.put(connName, connMock);
+ FieldSetter.setField(stateManager, stateManager.getClass().getDeclaredField("mdbcConnections"),
+ connMap);
+ TxCommitProgress txInfoMock = Mockito.mock(TxCommitProgress.class);
+ FieldSetter.setField(stateManager, stateManager.getClass().getDeclaredField("transactionInfo"),
+ txInfoMock);
+
+ DBInterface dbiMock = Mockito.mock(DBInterface.class);
+ Mockito.when(connMock.getDBInterface()).thenReturn(dbiMock);
+ Set<Range> allRanges = new HashSet<>();
+ allRanges.add(new Range("rangeToWarmup"));
+ allRanges.add(new Range("rangeToWarmup2"));
+ allRanges.add(new Range("eventualRange"));
+ Mockito.when(dbiMock.getSQLRangeSet()).thenReturn(allRanges);
+
+ List<Range> eventualRanges = new ArrayList<Range>();
+ eventualRanges.add(new Range("eventualRange"));
+ stateManager.setEventualRanges(eventualRanges);
+
+ assertEquals(2, stateManager.getRangesToWarmup().size());
+ assertTrue(stateManager.getRangesToWarmup().contains(new Range("rangeToWarmup")));
+ assertTrue(stateManager.getRangesToWarmup().contains(new Range("rangeToWarmup2")));
+ }
+
+ @Test
+ public void testSetWarmupRanges() {
+ Set<Range> warmupRanges = new HashSet<>();
+ warmupRanges.add(new Range("rangeToWarmup"));
+ warmupRanges.add(new Range("rangeToWarmup2"));
+ stateManager.setWarmupRanges(warmupRanges);
+ assertEquals(warmupRanges, stateManager.getRangesToWarmup());
+ }
+
+}
--- /dev/null
+mock-maker-inline
Code Review / music / mdbc.git / commitdiff