You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by ch...@apache.org on 2015/10/28 19:12:54 UTC
svn commit: r1711088 [3/4] - in /uima/sandbox/uima-ducc/trunk:
src/main/admin/ src/main/assembly/ src/main/resources/
uima-ducc-common/src/main/java/org/apache/uima/ducc/common/persistence/
uima-ducc-common/src/main/java/org/apache/uima/ducc/common/per...
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbManager.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbManager.java?rev=1711088&r1=1711087&r2=1711088&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbManager.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbManager.java Wed Oct 28 18:12:53 2015
@@ -20,180 +20,134 @@
package org.apache.uima.ducc.database;
import java.io.FileInputStream;
-import java.util.HashMap;
-import java.util.Map;
import java.util.Properties;
import org.apache.uima.ducc.common.utils.DuccLogger;
-import org.apache.uima.ducc.database.DbConstants.DbEdge;
-import org.apache.uima.ducc.database.DbConstants.DbVertex;
-
-import com.orientechnologies.orient.client.remote.OServerAdmin;
-import com.orientechnologies.orient.core.intent.OIntent;
-import com.tinkerpop.blueprints.impls.orient.OrientGraph;
-import com.tinkerpop.blueprints.impls.orient.OrientGraphFactory;
-import com.tinkerpop.blueprints.impls.orient.OrientGraphNoTx;
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Host;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SimpleStatement;
+
+/**
+ * Provide a common point for contacting the db, acquiring sessions/handles to it, managing the db,
+ * closing, etc.
+ */
public class DbManager
{
String dburl;
DuccLogger logger;
- OrientGraphFactory factory;
-
- Map<String, DbVertex> name_to_vertex = new HashMap<String, DbVertex>();
- Map<String, DbEdge> name_to_edge = new HashMap<String, DbEdge>();
-
- // private ODatabaseDocumentTx documentDb;
+ private Cluster cluster; // only one
+ private Session session; // only one - it's thread safe and manages a connection pool
public DbManager(String dburl, DuccLogger logger)
throws Exception
{
this.dburl = dburl;
this.logger = logger;
-
- for ( DbVertex o : DbVertex.values() ) {
- name_to_vertex.put(o.pname(), o);
- }
- for ( DbEdge o : DbEdge.values() ) {
- name_to_edge.put(o.pname(), o);
- }
-
}
-
- DbVertex vertexType(String v)
- {
- return name_to_vertex.get(v);
- }
-
- DbEdge EdgeType(String e)
- {
- return name_to_edge.get(e);
- }
-
boolean checkForDatabase()
throws Exception
{
String methodName = "checkForDatabase";
- String pw = dbPassword();
- OServerAdmin admin = null;
-
- boolean ret = true;
- try {
- admin = new OServerAdmin(dburl);
- admin.connect("root", pw); // connect to the server
-
- if ( ! admin.existsDatabase("plocal") ) {
- logger.info(methodName, null, "Database " + dburl + " does not exist.");
- ret = false;
- }
- } finally {
- if ( admin != null) admin.close();
- }
- return ret;
+ logger.warn(methodName, null, "Not yet implemented.");
+ return true;
}
public void drop()
throws Exception
{
- if ( dburl.startsWith("remote") ) {
- OServerAdmin admin = null;
- try {
- String pw = dbPassword();
- admin = new OServerAdmin(dburl);
- admin.connect("root", pw); // connect to the server
- admin.dropDatabase("plocal");
- admin.close();
- } finally {
- if ( admin != null ) admin.close();
- }
- } else {
- OrientGraphNoTx graphDb = factory.getNoTx(); // the graph instance
- if ( graphDb == null ) {
- throw new IllegalStateException("Cannot allocate graph instance for " + dburl);
- }
- graphDb.drop();
- }
+ String methodName = "drop";
+ logger.warn(methodName, null, "Drop is not implemented yet.");
}
public synchronized DbHandle open()
throws Exception
{
- OrientGraph graphDb = factory.getTx(); // the graph instance
- if ( graphDb == null ) {
- throw new IllegalStateException("Cannot allocate graph instance for " + dburl);
+ if ( session == null ) {
+ session = cluster.connect();
}
-
- graphDb.setUseLightweightEdges(true);
- return new DbHandle(this, graphDb);
+
+ return new DbHandle(this);
}
- public synchronized DbHandle openNoLog()
- throws Exception
+ public synchronized void init()
+ throws Exception
{
- OrientGraph graphDb = factory.getTx(); // the graph instance
- if ( graphDb == null ) {
- throw new IllegalStateException("Cannot allocate graph instance for " + dburl);
- }
+ String methodName = "init";
+
+ if ( cluster != null ) return; // already initialized
+
+ cluster = Cluster.builder()
+ .addContactPoint(dburl)
+ .build();
+
+ Metadata metadata = cluster.getMetadata();
+ logger.info(methodName, null, "Connected to cluster: %s\n", metadata.getClusterName());
- graphDb.setUseLightweightEdges(true);
- graphDb.getRawGraph().getTransaction().setUsingLog(false);
- return new DbHandle(this, graphDb);
+ for ( Host host : metadata.getAllHosts() ) {
+ logger.info(methodName, null, "Datatacenter: %s; Host: %s; Rack: %s\n", host.getDatacenter(), host.getAddress(), host.getRack());
+ }
}
-
- public synchronized DbHandle openNoTx()
- throws Exception
+ public synchronized void shutdown()
{
- OrientGraphNoTx graphDb = factory.getNoTx(); // the graph instance
- if ( graphDb == null ) {
- throw new IllegalStateException("Cannot allocate graph instance for " + dburl);
- }
- graphDb.setUseLightweightEdges(true);
- return new DbHandle(this, graphDb);
+ String methodName = "closeDatabase";
+ logger.info(methodName, null, "Closing the database.");
+ cluster.close();
+ cluster = null;
+ session = null;
}
-
- public synchronized void init()
- throws Exception
+ PreparedStatement prepare(String cql)
{
- String methodName = "init";
+ return session.prepare(cql);
+ }
- if ( factory != null ) return; // already initialized
+ void truncate(String table)
+ throws Exception
+ {
+ execute("TRUNCATE " + table);
+ }
- if ( ! dburl.startsWith("plocal") ) {
- // make sure the server is up if it's not a plocal db
- if ( !checkForDatabase() ) {
- throw new IllegalStateException("Database does not exist and must be created:" + dburl);
- }
+ String truncateText(String s)
+ {
+ String ret = s;
+ if ( ret.length() > 200 ) {
+ ret = s.substring(0, 200) + " ... ";
}
-
- factory = new OrientGraphFactory(dburl);
- if ( factory == null ) {
- throw new IllegalStateException("Cannot create graph factory for " + dburl);
- }
- logger.info(methodName, null, "Database is opened:", dburl);
- factory.setupPool(1,20);
+ return ret;
}
- public synchronized void declareIntent(OIntent intent)
+ ResultSet execute(String cql)
{
- factory.declareIntent(intent);
+ String methodName = "execute";
+ if ( logger.isDebug() ) {
+ logger.info(methodName, null, "EXECUTE CQL:", cql);
+ } else {
+ logger.info(methodName, null, "EXECUTE CQL:", truncateText(cql));
+ }
+ return session.execute(cql);
}
- public synchronized void shutdown()
+ ResultSet execute(BoundStatement s)
{
- String methodName = "closeDatabase";
- logger.info(methodName, null, "Closing the database.");
- if ( factory != null ) {
- // closes all pooled instances and stops the factory
- factory.close();
- factory = null;
- }
+ return session.execute(s);
}
+ ResultSet execute(SimpleStatement s)
+ {
+ String methodName = "execute";
+ logger.info(methodName, null, "EXECUTE STATEMENT:", truncateText(s.getQueryString()));
+ return session.execute(s);
+ }
static String dbPassword()
throws Exception
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbUtil.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbUtil.java?rev=1711088&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbUtil.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbUtil.java Wed Oct 28 18:12:53 2015
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+*/
+
+package org.apache.uima.ducc.database;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.uima.ducc.common.persistence.IDbProperty;
+import org.apache.uima.ducc.common.persistence.IDbProperty.Type;
+
+/**
+ * Static common helper methods.
+ *
+ * Not public at this point, would prefer to encapsulate all this entirely in DB.
+ */
+class DbUtil
+{
+ static String mkSchema(IDbProperty[] props)
+ throws Exception
+ {
+ List<String> parts = new ArrayList<String>();
+ List<String> primaries = new ArrayList<String>();
+ for (IDbProperty n : props ) {
+ if ( n.isMeta() ) continue;
+ String s = n.columnName() + " " + typeToString(n.type());
+ if ( n.isPrimaryKey() ) {
+ primaries.add(n.columnName());
+ }
+ parts.add(s);
+ }
+ if ( primaries.size() == 0 ) {
+ throw new IllegalArgumentException("Schema properties must declare at least one primary key.");
+ }
+ StringBuffer buf = new StringBuffer();
+ for ( String p : parts ) {
+ buf.append(p);
+ buf.append(",");
+ }
+ int ncommas = primaries.size() - 1;
+ int c = 0;
+ buf.append(" PRIMARY KEY(");
+ for ( String s : primaries ) {
+ buf.append(s);
+ if ( c++ < ncommas ) {
+ buf.append(",");
+ }
+ }
+
+ buf.append(")");
+ return buf.toString();
+ }
+
+ static String mkFields(StringBuffer buf, String[] fields)
+ {
+ int max = fields.length - 1;
+ int current = 0;
+ buf.append("(");
+ for (String s : fields) {
+ buf.append(s);
+ if ( current++ < max) buf.append(", ");
+ }
+ buf.append(")");
+ return buf.toString();
+
+ }
+
+ /**
+ * Generate a CREATE TABLE statement from the incoming fields. The preparer of the
+ * fields must qualify any fields in advance e.g. with types, key attributes, etc.
+ *
+ * @param tableName This is the name of the table to create.
+ * @param fields This is a string array of fields to generate the statement from.
+ *
+ * @return A string of valid SQL / CQL used to create the table.
+ */
+ static String mkTableCreate(String tableName, String[] fields)
+ {
+ int max = fields.length - 1;
+ int current = 0;
+ StringBuffer buf = new StringBuffer("CREATE TABLE IF NOT EXISTS ");
+ buf.append(tableName);
+ buf.append(" (");
+ for (String s : fields) {
+ buf.append(s);
+ if ( current++ < max) buf.append(", ");
+ }
+ buf.append(")");
+ return buf.toString();
+ }
+
+ static String mkInsert(String tableName, Map<? extends IDbProperty, Object> props)
+ {
+ int max = props.size() - 1;
+ int current = 0;
+ StringBuffer buf = new StringBuffer("INSERT INTO ");
+ buf.append(tableName);
+ buf.append("(");
+
+ StringBuffer vals = new StringBuffer(") VALUES (");
+
+ for ( IDbProperty ok : props.keySet() ) {
+
+ String k = ok.columnName();
+ buf.append(k);
+ vals.append(rep(ok.type(), props.get(ok)));
+
+ if ( current++ < max ) {
+ buf.append(",");
+ vals.append(",");
+ }
+ }
+ buf.append(vals.toString());
+ buf.append(")");
+
+ return buf.toString();
+ }
+
+ static String mkInsert(String tableName, Object key, Object keyval, Map<? extends IDbProperty, Object> props)
+ {
+ int max = props.size() + 1;
+ int current = 0;
+ StringBuffer buf = new StringBuffer("INSERT INTO ");
+ buf.append(tableName);
+ buf.append("(");
+
+ StringBuffer vals = new StringBuffer(") VALUES (");
+
+ buf.append(key.toString());
+ buf.append(",");
+ vals.append(keyval.toString());
+ vals.append(",");
+
+ for ( IDbProperty ok : props.keySet() ) {
+
+ String k = ok.columnName();
+ buf.append(k);
+ vals.append(rep(ok.type(), props.get(ok)));
+
+ if ( current++ < max ) {
+ buf.append(",");
+ vals.append(",");
+ }
+ }
+ buf.append(vals.toString());
+ buf.append(")");
+
+ return buf.toString();
+
+ }
+
+ static String mkUpdate(String table, String key, Object... props)
+ {
+ int len = props.length;
+ StringBuffer buf = new StringBuffer("UPDATE ");
+ buf.append(table);
+ buf.append(" SET ");
+
+ for ( int i = 0; i < len; i+=2) {
+ IDbProperty prop = (IDbProperty) props[i];
+ if ( !prop.isPrimaryKey()) { // not allowed to update this
+ // we allow it in 'props' so callers can
+ // simply call update and expect the right
+ // thing to happen
+
+ buf.append(prop.columnName());
+ buf.append("=");
+ buf.append(rep(prop.type(), props[i+1]));
+ if ( i + 2 < len ) {
+ buf.append(",");
+ }
+ }
+ }
+ buf.append(" WHERE ");
+ buf.append(key);
+ return buf.toString();
+ }
+
+ /**
+ * Return the correct representation for CQL update, of val, for the indicated type, for this database.
+ */
+ static String rep(Type t, Object val)
+ {
+ switch ( t ) {
+ case String:
+ return "'" + val.toString() + "'";
+ default:
+ return val.toString();
+ }
+
+ }
+
+ /**
+ * Convert our generic "type" to the right name for this db implementation.
+ * We could make Type a magic enum but I want to hide DB specifics, in particular,
+ * how this database names various java types.
+ */
+ static String typeToString(Type t)
+ {
+ switch ( t ) {
+ case Blob:
+ return "blob";
+ case String:
+ return "text";
+ case Boolean:
+ return "boolean";
+ case Integer:
+ return "int";
+ case Long:
+ return "bigint";
+ case Double:
+ return "double";
+ case UUID:
+ return "uuid";
+ }
+ throw new IllegalArgumentException("Unrecognized type for schema: " + t);
+ }
+
+}
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbVerify.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbVerify.java?rev=1711088&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbVerify.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/DbVerify.java Wed Oct 28 18:12:53 2015
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.uima.ducc.database;
+
+import java.io.ByteArrayInputStream;
+import java.io.ObjectInputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import org.apache.uima.ducc.common.persistence.services.StateServicesDirectory;
+import org.apache.uima.ducc.common.persistence.services.StateServicesSet;
+import org.apache.uima.ducc.common.utils.DuccLogger;
+import org.apache.uima.ducc.common.utils.id.DuccId;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.SimpleStatement;
+
+/**
+ * Toy orientdb loader to load a historydb from ducc history
+ */
+
+public class DbVerify
+{
+ DuccLogger logger = DuccLogger.getLogger(DbLoader.class, "DBVERIFY");
+ String DUCC_HOME;
+
+ DbManager dbManager = null;
+ long total_bytes = 0;
+
+ public DbVerify()
+ throws Exception
+ {
+ //String methodName = "<ctr>";
+ DUCC_HOME = System.getProperty("DUCC_HOME");
+ if ( DUCC_HOME == null ) {
+ System.out.println("System proprety -DDUCC_HOME must be set.");
+ System.exit(1);
+ }
+ }
+
+
+ void verify(String table)
+ throws Exception
+ {
+ String methodName = "verify";
+ DbHandle h = dbManager.open();
+ SimpleStatement s = new SimpleStatement("SELECT * from " + table);
+ //SimpleStatement s = new SimpleStatement("SELECT * from " + table + " LIMIT 10"); // for test and debug
+ logger.info(methodName, null, "Fetch size", s.getFetchSize());
+ s.setFetchSize(100);
+ long now = System.currentTimeMillis();
+
+ int counter = 0;
+ int nbytes = 0;
+ try {
+ ResultSet rs = h.execute(s);
+ for ( Row r : rs ) {
+ counter++;
+ ByteBuffer b = r.getBytes("work");
+ byte[] bytes = b.array();
+ nbytes += bytes.length;
+ total_bytes += bytes.length;;
+
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ Object o = ois.readObject();
+ ois.close();
+ DuccId did = new DuccId(r.getLong("ducc_dbid"));
+
+ logger.info(methodName, did, "found object class", o.getClass().getName(), "of type", r.getString("type"), "in table", table, "of size", bytes.length);
+ }
+ } catch (Exception e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+
+ logger.info(methodName, null, "Found", counter, "results. Total bytes", nbytes);
+ logger.info(methodName, null, "Total time for", table, System.currentTimeMillis() - now);
+ }
+
+ void verifyServices()
+ throws Exception
+ {
+ String methodName = "verify";
+ int live = 0;
+ int archived = 0;
+ StateServicesDb sdb = new StateServicesDb();
+ sdb.init(logger,dbManager);
+
+ StateServicesDirectory ssd = sdb.fetchServices(true); // first the archived stuff
+ Map<Long, StateServicesSet> svcmap = ssd.getMap();
+ for ( Long id : svcmap.keySet() ) {
+ DuccId did = new DuccId(id);
+ archived++;
+ logger.info(methodName, did, "Found an archived service.");
+ }
+
+ ssd = sdb.fetchServices(false); // first the archived stuff
+ svcmap = ssd.getMap();
+ for ( Long id : svcmap.keySet() ) {
+ DuccId did = new DuccId(id);
+ logger.info(methodName, did, "Found a live service.");
+ live++;
+ }
+ logger.info(methodName, null, "Found", live, "live services and", archived, "archived services.");
+
+ }
+
+ void run()
+ throws Exception
+ {
+ String methodName = "run";
+ long now = System.currentTimeMillis();
+ String state_url = "bluej538";
+ try {
+ dbManager = new DbManager(state_url, logger);
+ dbManager.init();
+
+ verifyServices();
+
+ if ( false ) verify("ducc.res_history");
+ if ( false ) verify("ducc.svc_history");
+ if ( false ) verify("ducc.job_history");
+
+ } finally {
+ dbManager.shutdown();
+ }
+ logger.info(methodName, null, "Read", total_bytes, "bytes in", System.currentTimeMillis() - now, "MS");
+ }
+
+
+ public static void main(String[] args)
+ {
+ DbVerify v = null;
+ try {
+ v = new DbVerify();
+ v.run();
+ } catch ( Exception e ) {
+ e.printStackTrace();
+ }
+ }
+}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/HistoryManagerDb.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/HistoryManagerDb.java?rev=1711088&r1=1711087&r2=1711088&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/HistoryManagerDb.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/HistoryManagerDb.java Wed Oct 28 18:12:53 2015
@@ -18,52 +18,33 @@
*/
package org.apache.uima.ducc.database;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.Set;
-import org.apache.uima.ducc.common.SizeBytes;
-import org.apache.uima.ducc.common.main.DuccService;
+import org.apache.uima.ducc.common.Pair;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.id.DuccId;
-import org.apache.uima.ducc.database.DbConstants.DbCategory;
-import org.apache.uima.ducc.database.DbConstants.DbEdge;
-import org.apache.uima.ducc.database.DbConstants.DbVertex;
-import org.apache.uima.ducc.transport.event.common.ADuccWork;
-import org.apache.uima.ducc.transport.event.common.DuccProcess;
-import org.apache.uima.ducc.transport.event.common.DuccProcessMap;
-import org.apache.uima.ducc.transport.event.common.DuccReservation;
-import org.apache.uima.ducc.transport.event.common.DuccWorkJob;
import org.apache.uima.ducc.transport.event.common.DuccWorkMap;
-import org.apache.uima.ducc.transport.event.common.DuccWorkPopDriver;
import org.apache.uima.ducc.transport.event.common.DuccWorkReservation;
-import org.apache.uima.ducc.transport.event.common.IDuccCompletionType.JobCompletionType;
-import org.apache.uima.ducc.transport.event.common.IDuccCompletionType.ReservationCompletionType;
-import org.apache.uima.ducc.transport.event.common.IDuccPerWorkItemStatistics;
-import org.apache.uima.ducc.transport.event.common.IDuccProcess;
-import org.apache.uima.ducc.transport.event.common.IDuccProcessMap;
-import org.apache.uima.ducc.transport.event.common.IDuccReservation;
-import org.apache.uima.ducc.transport.event.common.IDuccReservationMap;
-import org.apache.uima.ducc.transport.event.common.IDuccState.JobState;
-import org.apache.uima.ducc.transport.event.common.IDuccState.ReservationState;
+import org.apache.uima.ducc.transport.event.common.IDuccTypes.DuccType;
import org.apache.uima.ducc.transport.event.common.IDuccWork;
import org.apache.uima.ducc.transport.event.common.IDuccWorkJob;
import org.apache.uima.ducc.transport.event.common.IDuccWorkReservation;
import org.apache.uima.ducc.transport.event.common.IDuccWorkService;
-import org.apache.uima.ducc.transport.event.common.JdReservationBean;
import org.apache.uima.ducc.transport.event.common.history.IHistoryPersistenceManager;
-import com.google.gson.Gson;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.SimpleStatement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
-import com.orientechnologies.orient.core.intent.OIntentMassiveInsert;
-import com.orientechnologies.orient.core.record.impl.ODocument;
-import com.tinkerpop.blueprints.Direction;
-import com.tinkerpop.blueprints.Edge;
-import com.tinkerpop.blueprints.Vertex;
-import com.tinkerpop.blueprints.impls.orient.OrientEdge;
-import com.tinkerpop.blueprints.impls.orient.OrientVertex;
public class HistoryManagerDb
@@ -72,265 +53,225 @@ public class HistoryManagerDb
private DuccLogger logger = null;
- private String dburl;
private DbManager dbManager;
+
+ PreparedStatement jobPrepare = null;
+ PreparedStatement reservationPrepare = null;
+ PreparedStatement servicePrepare = null;
+ PreparedStatement ckptPrepare = null;
+ static final String JOB_TABLE = "ducc." + OrWorkProps.JOB_TABLE.pname();
+ static final String RES_TABLE = "ducc." + OrWorkProps.RESERVATION_TABLE.pname();
+ static final String SVC_TABLE = "ducc." + OrWorkProps.SERVICE_TABLE.pname();
+ static final String CKPT_TABLE = "ducc." + OrCkptProps.CKPT_TABLE.pname();
public HistoryManagerDb()
{
- this(DuccService.getDuccLogger(HistoryManagerDb.class.getName()));
}
-
- public HistoryManagerDb(DuccLogger logger)
- {
-
- this.logger = logger;
- dburl = System.getProperty("ducc.state.database.url");
+
+
+ private boolean init(String dburl, DbManager dbm)
+ throws Exception
+ {
+ String methodName = "init";
+ boolean ret = true;
+ logger.info(methodName, null, "Initializing OR persistence over the database");
try {
- dbManager = new DbManager(dburl, logger);
- dbManager.init();
- // TODO TODO
- dbManager.declareIntent(new OIntentMassiveInsert());
- logger.warn("<CTR>.HistoryManagerDb", null, "****MUST FIX DECLARE INTENT****");
+ if ( dbm != null ) {
+ this.dbManager = dbm;
+ } else {
+ dbManager = new DbManager(dburl, logger);
+ dbManager.init();
+ }
+
+ // prepare some statements
+ DbHandle h = dbManager.open();
+ jobPrepare = h.prepare("INSERT INTO " + JOB_TABLE + " (ducc_dbid, type, history, work) VALUES (?, ?, ?, ?) IF NOT EXISTS;");
+ reservationPrepare = h.prepare("INSERT INTO " + RES_TABLE + " (ducc_dbid, type, history, work) VALUES (?, ?, ?, ?) IF NOT EXISTS;");
+ servicePrepare = h.prepare("INSERT INTO " + SVC_TABLE + " (ducc_dbid, type, history, work) VALUES (?, ?, ?, ?) IF NOT EXISTS;");
+ ckptPrepare = h.prepare("INSERT INTO " + CKPT_TABLE + " (id, work, p2jmap) VALUES (?, ?, ?);");
+
} catch ( Exception e ) {
- logger.error("HisstoryManagerDb", null, "Cannot open the history database:", e);
- }
+ logger.error(methodName, null, "Cannot open the history database:", e);
+ throw e;
+ }
+ return ret;
}
- public void setLogger(DuccLogger logger)
+ public boolean init(DuccLogger logger)
+ throws Exception
{
this.logger = logger;
+ String historyUrl = System.getProperty("ducc.state.database.url");
+ return init(historyUrl, null);
+ }
+
+ // package only, for the loader
+ boolean init(DuccLogger logger, DbManager dbManager)
+ throws Exception
+ {
+ this.logger = logger;
+ String stateUrl = System.getProperty("ducc.state.database.url");
+ return init(stateUrl, dbManager);
}
- // ----------------------------------------------------------------------------------------------------
- // Jobs section
/**
- * Common code to save a job in an open handle. Caller will commit or fail as needed.
+ * Schema gen. Do anything you want to make the schema, but notice that DbUtil has a few convenience methods if
+ * you want to define your schema in a magic enum.
*/
- void saveJobNoCommit(DbHandle h, IDuccWorkJob j, DbVertex type, DbCategory dbcat)
- throws Exception
+ static ArrayList<SimpleStatement> mkSchema(String tablename)
+ throws Exception
{
- String methodName = "saveJobNoCommit";
- Long nowP = System.currentTimeMillis();
- // Nuke the command lines
- DuccWorkPopDriver driver = j.getDriver();
- //ICommandLine driverCl = null;
- IDuccProcessMap jdProcessMap = null;
-
- int size = 0;
-
- if ( driver != null ) {
- //driverCl = driver.getCommandLine();
- //driver.setCommandLine(null);
- jdProcessMap = driver.getProcessMap();
- driver.setProcessMap(null);
- }
+ ArrayList<SimpleStatement> ret = new ArrayList<SimpleStatement>();
- //ICommandLine jobCl = j.getCommandLine();
- //j.setCommandLine(null);
+ StringBuffer buf = new StringBuffer("CREATE TABLE IF NOT EXISTS " + tablename + " (");
+ buf.append(DbUtil.mkSchema(OrWorkProps.values()));
+ buf.append(")");
+ buf.append("WITH CLUSTERING ORDER BY (ducc_dbid desc)");
+
+ ret.add(new SimpleStatement(buf.toString()));
+ ret.add(new SimpleStatement("CREATE INDEX IF NOT EXISTS ON " + tablename + "(ducc_dbid)"));
+ ret.add(new SimpleStatement("CREATE INDEX IF NOT EXISTS ON " + tablename + "(history)"));
- IDuccPerWorkItemStatistics stats = j.getSchedulingInfo().getPerWorkItemStatistics();
+ return ret;
+ }
- if ( stats != null ) {
- if (Double.isNaN(stats.getStandardDeviation()) ) {
- stats.setStandardDeviation(0.0);
- }
- }
+ static ArrayList<SimpleStatement> mkSchema()
+ throws Exception
+ {
+ ArrayList<SimpleStatement> ret = new ArrayList<SimpleStatement>();
- // Pull process map so we can put processes in their own records
- IDuccProcessMap processMap = j.getProcessMap();
- j.setProcessMap(null);
-
- Gson g = DbHandle.mkGsonForJob();
-
- String dbJob = g.toJson(j);
- size += dbJob.length();
-
- // Must repair these things because OR continues to use the job after it has been
- // written to history.
- j.setProcessMap(processMap);
- //j.setCommandLine(jobCl);
- if ( driver != null ) {
- //driver.setCommandLine(driverCl);
- driver.setProcessMap(jdProcessMap);
- }
-
- OrientVertex savedJob = h.saveObject(type, j.getDuccId().getFriendly(), dbJob, dbcat);
-
- List<OrientVertex> savedJPs = new ArrayList<OrientVertex>();
- List<OrientVertex> savedJDs = new ArrayList<OrientVertex>();
- for (DuccId did : processMap.keySet()) {
- Long pid = did.getFriendly();
-
- IDuccProcess p = processMap.get(did);
- String proc = g.toJson(p);
- size += proc.length();
-
- savedJPs.add(h.saveObject(DbVertex.Process, pid, proc, dbcat));
- // logger.info(methodName, p.getDuccId(), "2 ----------> Time to save process", System.currentTimeMillis() - nowP);
-
- }
-
- if ( driver != null ) {
- for (DuccId did : jdProcessMap.keySet()) {
- Long pid = did.getFriendly();
-
- IDuccProcess p = jdProcessMap.get(did);
- String proc = g.toJson(p);
- size += proc.length();
-
- savedJDs.add(h.saveObject(DbVertex.Process, pid, proc, dbcat));
- // logger.info(methodName, p.getDuccId(), "2 ----------> Time to save process", System.currentTimeMillis() - nowP);
-
- }
- h.addEdges(savedJob, savedJDs, DbEdge.JdProcess);
- }
-
- h.addEdges(savedJob, savedJPs, DbEdge.JpProcess);
+ ret.addAll(mkSchema(JOB_TABLE));
+ ret.addAll(mkSchema(RES_TABLE));
+ ret.addAll(mkSchema(SVC_TABLE));
+
+ StringBuffer buf = new StringBuffer("CREATE TABLE IF NOT EXISTS " + CKPT_TABLE + " (");
+ buf.append(DbUtil.mkSchema(OrCkptProps.values()));
+ buf.append(")");
+ ret.add(new SimpleStatement(buf.toString()));
- logger.info(methodName, j.getDuccId(), "----------> Time to save job", System.currentTimeMillis() - nowP, "json size", size, "nprocesses", processMap.size());
-
+ return ret;
}
-
- private void saveJobInternal(IDuccWorkJob j, DbVertex type, boolean safe, DbCategory dbcat)
- throws Exception
- {
-
- // It seems that services instances are represented as jobs without drivers. So we use
- // the common code here, passing in the vertex type, for both jobs and services.
- String methodName = "saveJob";
- logger.info(methodName, j.getDuccId(), "Saving: type:", type.pname(), "safe:", safe, "DbCategory:", dbcat);
+ // ----------------------------------------------------------------------------------------------------
+ // Jobs section
- Long id = j.getDuccId().getFriendly();
- DbHandle h = null;
- try {
+ void saveWork(PreparedStatement s, IDuccWork w, boolean isHistory)
+ throws Exception
+ {
+ String methodName = "saveWork";
+ Long nowP = System.currentTimeMillis();
+ String type = null;
+ if ( w instanceof IDuccWorkJob ) {
+ type = "job";
+ } else if ( w instanceof IDuccWorkReservation ) {
+ type = "reservation";
+ } else if ( w instanceof IDuccWorkService ) {
+ type = "service";
+ } else {
+ throw new IllegalArgumentException("Improper object passed to saveWork");
+ }
+
+ logger.info(methodName, w.getDuccId(), "-------- saving " + type);
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(baos);
+ out.writeObject(w);
+ out.close();
+ byte[] bytes = baos.toByteArray();
+ ByteBuffer buf = ByteBuffer.wrap(bytes);
- if ( safe ) {
- h = dbManager.open();
- } else {
- h = dbManager.openNoLog();
- }
+ DbHandle h = dbManager.open();
+ h.saveObject(jobPrepare, w.getDuccId().getFriendly(), type, isHistory, buf);
- if ( safe && h.thingInDatabase(id, type, dbcat) ) {
- logger.warn(methodName, j.getDuccId(), "Not overwriting saved job.");
- h.close();
- return;
- }
- } catch ( Exception e ) {
- if ( h != null ) h.close();
- throw e;
- }
-
- try {
- saveJobNoCommit(h, j, type, dbcat);
- } catch ( Exception e ) {
- h.rollback();
- logger.error(methodName, j.getDuccId(), "Cannot store job", e);
- throw e;
- } finally {
- Long nowP = System.currentTimeMillis();
- h.commit();
- logger.info(methodName, j.getDuccId(), "Time to commit", System.currentTimeMillis() - nowP);
- h.close();
- }
- }
+ logger.info(methodName, w.getDuccId(), "----------> Time to save", type, ":", System.currentTimeMillis() - nowP, "Size:", bytes.length, "bytes.");
+ }
/**
- * For use by the loader, load it without the existence check; the assumption this is a first-time load
- * and the check isn't needed. This saves history only.
+ * Part of history management, recover ths indicated job from history.
*/
- public void saveJobUnsafe(IDuccWorkJob j)
+ <T> T restoreWork(Class<T> cl, String tablename, long friendly_id)
throws Exception
{
- saveJobInternal(j, DbVertex.Job, false, DbCategory.History);
- }
+ String methodName = "restoreWork";
+ T ret = null;
+ DbHandle h = null;
+ h = dbManager.open();
+ String cql = "SELECT WORK FROM " + tablename + " WHERE DUCC_DBID=" + Long.toString(friendly_id);
+ ResultSet rs = h.execute(cql);
+ for ( Row r : rs ) {
+ logger.info(methodName, null, "----- Restoring", friendly_id);
+ ByteBuffer bbWork = r.getBytes("work");
+
+ byte[] workbytes = bbWork.array();
+ ByteArrayInputStream bais = new ByteArrayInputStream(workbytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ ret= (T) ois.readObject();
+ ois.close();
+ }
+
+ return ret;
+ }
+
/**
- * For use by normal operation: forces an existence check. This saves history only.
+ * Part of history management, recover ths indicated jobs from history.
+ *
+ * Reminder to self, we need to pass Clas<T> cl so compiler can infer T.
*/
- public void saveJob(IDuccWorkJob j)
- throws Exception
- {
- saveJobInternal(j, DbVertex.Job, true, DbCategory.History);
- }
-
-
- private IDuccWorkJob restoreJobInternal(DbHandle h, OrientVertex v)
+ public <T> ArrayList<T> restoreSeveralThings(Class<T> cl, String tablename, long max)
throws Exception
{
- IDuccWorkJob j = null;
-
- ODocument d = v.getRecord();
- String json = d.toJSON();
- JsonObject jo = mkJsonObject(json);
+ String methodName = "restoreSeveralThings";
- Gson g = DbHandle.mkGsonForJob();
- j = g.fromJson(jo, DuccWorkJob.class);
-
- // System.out.println(g.toJson(jo));
-
- IDuccProcessMap pm = j.getProcessMap(); // seems to get set by default when job is recovered
- Iterable<Edge> ed = v.getEdges(Direction.OUT, DbEdge.JpProcess.pname());
- for ( Edge e : ed ) {
- OrientEdge oe = (OrientEdge) e;
- OrientVertex ov = oe.getVertex(Direction.IN);
-
- ODocument pd = ov.getRecord();
- String pjson = pd.toJSON();
-
- IDuccProcess pe = g.fromJson(pjson, DuccProcess.class);
- pm.addProcess(pe);
- }
+ ArrayList<T> ret = new ArrayList<T>();
+ DbHandle h = dbManager.open();
+ SimpleStatement s = new SimpleStatement("SELECT * from " + tablename + " limit " + max);
+ s.setFetchSize(100);
+ long now = System.currentTimeMillis();
- DuccWorkPopDriver driver = j.getDriver();
- if ( driver != null ) {
- pm = new DuccProcessMap();
- driver.setProcessMap(pm); // seems NOT to get set when driver is reconstituted
- ed = v.getEdges(Direction.OUT, DbEdge.JdProcess.pname());
- for ( Edge e : ed ) {
- OrientEdge oe = (OrientEdge) e;
- OrientVertex ov = oe.getVertex(Direction.IN);
-
- ODocument pd = ov.getRecord();
- String pjson = pd.toJSON();
-
- IDuccProcess pe = g.fromJson(pjson, DuccProcess.class);
- pm.addProcess(pe);
+ try {
+ int count = 0;
+ int nbytes = 0;
+ ResultSet rs = h.execute(s);
+ for ( Row r : rs ) {
+ count++;
+ ByteBuffer b = r.getBytes("work");
+ byte[] workbytes = b.array();
+ nbytes += workbytes.length;
+
+ ByteArrayInputStream bais = new ByteArrayInputStream(workbytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ ret.add( (T) ois.readObject());
+ ois.close();
+ count++;
}
- }
+
+ logger.info(methodName, null, "Found", count, "results. Total bytes", nbytes, "Time:", System.currentTimeMillis() - now);
+ } catch (Exception e) {
+ logger.error(methodName, null, "Error fetching history:", e);
+ }
+ return ret;
+ }
- // Now must hack around becase this 'n that and JSON can't work out some things
- String ct = (String) ((ADuccWork)j).getCompletionTypeObject();
- j.setCompletionType(JobCompletionType.valueOf(ct));
- String so = (String) ((ADuccWork)j).getStateObject();
- j.setJobState(JobState.valueOf(so));
-
- return j;
+ /**
+ * For use by normal operation: forces an existence check. This saves history only.
+ */
+ public void saveJob(IDuccWorkJob j)
+ throws Exception
+ {
+ saveWork(jobPrepare, j, true);
}
+
/**
* Part of history management, recover ths indicated job from history.
*/
public IDuccWorkJob restoreJob(long friendly_id)
throws Exception
{
- DuccWorkJob ret = null;
- DbHandle h = null;
- try {
- h = dbManager.open();
- Iterable<Vertex> q = h.select("SELECT * FROM " + DbVertex.Job.pname() + " WHERE ducc_dbid=" + friendly_id +
- " AND " + DbConstants.DUCC_DBCAT + "='" + DbCategory.History.pname() + "'");
- for ( Vertex v : q ) {
- // There's only 1 unless db is broken.
- return restoreJobInternal(h, (OrientVertex) v);
- }
- } finally {
- h.close();
- }
-
- return ret;
+ return (IDuccWorkJob) restoreWork(IDuccWorkJob.class, JOB_TABLE, friendly_id);
}
/**
@@ -339,22 +280,7 @@ public class HistoryManagerDb
public ArrayList<IDuccWorkJob> restoreJobs(long max)
throws Exception
{
- ArrayList<IDuccWorkJob> ret = new ArrayList<IDuccWorkJob>();
- DbHandle h = null;
- try {
- h = dbManager.open();
- Iterable<Vertex> q = h.select("SELECT * FROM " + DbVertex.Job.pname() +
- " where " + DbConstants.DUCC_DBCAT +"='" + DbCategory.History.pname() +
- "' ORDER BY ducc_dbid DESC LIMIT "+ max);
- for ( Vertex v : q ) {
- IDuccWorkJob j = restoreJobInternal(h, (OrientVertex) v);
- ret.add(j);
- }
- } finally {
- h.close();
- }
-
- return ret;
+ return restoreSeveralThings(IDuccWorkJob.class, JOB_TABLE, max);
}
// End of jobs section
// ----------------------------------------------------------------------------------------------------
@@ -363,179 +289,21 @@ public class HistoryManagerDb
// ----------------------------------------------------------------------------------------------------
// Reservations section
- private void saveReservationNoCommit(DbHandle h, IDuccWorkReservation r, DbCategory dbcat)
- throws Exception
- {
- String methodName = "saveReservationNoCommit";
- long now = System.currentTimeMillis();
-
- List<JdReservationBean> l = r.getJdReservationBeanList();
- if ( l != null ) {
- for (JdReservationBean b : l ) {
- ConcurrentHashMap<DuccId, SizeBytes> map = b.getMap();
- for ( DuccId k : map.keySet() ) {
- logger.info(methodName, null, "SAVE ===> " + k.getFriendly() + " " + k.getUnique() + " : " + map.get(k));
- }
- }
- }
-
-
-
-
- Long id = r.getDuccId().getFriendly();
- logger.info(methodName, r.getDuccId(), "Saving.");
-
- // Nuke the command lines
-
- IDuccReservationMap resmap = r.getReservationMap();
- r.setReservationMap(null);
-
- Gson g = DbHandle.mkGsonForJob();
-
- String dbres = g.toJson(r);
- // logger.info(methodName, null, "------------------- Reservation JSON: " + dbres);
-
- // Must repair these things because OR continues to use the job after it has been
- // written to history.
- r.setReservationMap(resmap);
-
- OrientVertex savedRes = h.saveObject(DbVertex.Reservation, id, dbres, dbcat);
-
- List<OrientVertex> savedHosts = new ArrayList<OrientVertex>();
- for (DuccId did : resmap.keySet()) {
- Long pid = did.getFriendly();
-
- IDuccReservation p = resmap.get(did);
- String proc = g.toJson(p);
-
- savedHosts.add(h.saveObject(DbVertex.Process, pid, proc, dbcat));
- // logger.info(methodName, p.getDuccId(), "2 ----------> Time to save process", System.currentTimeMillis() - nowP);
-
- }
-
- h.addEdges(savedRes, savedHosts, DbEdge.JpProcess);
- logger.info(methodName, r.getDuccId(), "----------> Total reservation save time:", System.currentTimeMillis() - now, "nPE", resmap.size());
-
- }
-
- private void saveReservationInternal(IDuccWorkReservation r, boolean safe, DbCategory dbcat)
- throws Exception
- {
- String methodName = "saveReservation";
-
- Long id = r.getDuccId().getFriendly();
- DbHandle h = null;
- try {
- if ( safe ) {
- h = dbManager.open();
- } else {
- h = dbManager.openNoTx();
- }
- if ( safe && h.thingInDatabase(id, DbVertex.Reservation, dbcat) ) {
- h.close();
- return;
- }
- } catch ( Exception e ) {
- logger.warn(methodName, r.getDuccId(), e);
- h.close();
- return;
- }
-
- try {
- saveReservationNoCommit(h, r, dbcat);
- } catch ( Exception e ) {
- h.rollback();
- logger.error(methodName, r.getDuccId(), "Cannot store reservation:", e);
- } finally {
- h.commit();
- h.close();
- }
-
- }
-
// Save to history only
public void saveReservation(IDuccWorkReservation r)
throws Exception
{
- saveReservationInternal(r, true, DbCategory.History);
- }
-
- public void saveReservationUnsafe(IDuccWorkReservation r)
- throws Exception
- {
- saveReservationInternal(r, false, DbCategory.History);
+ saveWork(reservationPrepare, r, true);
}
- private IDuccWorkReservation restoreReservationInternal(DbHandle h, OrientVertex v)
- throws Exception
- {
- // String methodName = "restoreReservationInternal";
- IDuccWorkReservation r = null;
-
- ODocument d = v.getRecord();
- String json = d.toJSON();
- JsonObject jo = mkJsonObject(json);
-
- Gson g = DbHandle.mkGsonForJob();
- // logger.info(methodName, null, g.toJson(jo));
-
- r = g.fromJson(jo, DuccWorkReservation.class);
-
- //List<JdReservationBean> l = r.getJdReservationBeanList();
- //if ( l != null ) {
- //for (JdReservationBean b : l ) {
- //ConcurrentHashMap<DuccId, SizeBytes> map = b.getMap();
- //for ( DuccId k : map.keySet() ) {
- // logger.info(methodName, null, "REST ===> " + k.getFriendly() + " " + k.getUnique() + " : " + map.get(k));
- //}
- //}
- //}
-
- IDuccReservationMap rm = r.getReservationMap(); // seems to get set by default when job is recovered
- Iterable<Edge> ed = v.getEdges(Direction.OUT, DbEdge.JpProcess.pname());
- for ( Edge e : ed ) {
- OrientEdge oe = (OrientEdge) e;
- OrientVertex ov = oe.getVertex(Direction.IN);
-
- ODocument pd = ov.getRecord();
- String pjson = pd.toJSON();
-
- IDuccReservation rr = g.fromJson(pjson, DuccReservation.class);
- rm.addReservation(rr);
- }
-
- // Now must hack around becase this 'n that and JSON can't work out some things
- String ct = (String) ((ADuccWork)r).getCompletionTypeObject();
- r.setCompletionType(ReservationCompletionType.valueOf(ct));
-
- String so = (String) ((ADuccWork)r).getStateObject();
- r.setReservationState(ReservationState.valueOf(so));
-
- return r;
- }
-
/**
* Part of history management, recover ths indicated reservation from history.
*/
public IDuccWorkReservation restoreReservation(long duccid)
throws Exception
{
- DuccWorkReservation ret = null;
- DbHandle h = null;
- try {
- h = dbManager.open();
- Iterable<Vertex> q = h.select("SELECT * FROM " + DbVertex.Reservation.pname() + " WHERE ducc_dbid=" + duccid +
- " AND " + DbConstants.DUCC_DBCAT +"='" + DbCategory.History.pname() + "'");
- for ( Vertex v : q ) {
- // There's only 1 unless db is broken.
- return restoreReservationInternal(h, (OrientVertex) v);
- }
- } finally {
- h.close();
- }
-
- return ret;
- }
+ return (IDuccWorkReservation) restoreWork(IDuccWorkReservation.class, RES_TABLE, duccid);
+ }
/**
* Part of history management, recover ths indicated reservations from history.
@@ -543,44 +311,20 @@ public class HistoryManagerDb
public ArrayList<IDuccWorkReservation> restoreReservations(long max)
throws Exception
{
- ArrayList<IDuccWorkReservation> ret = new ArrayList<IDuccWorkReservation>();
- DbHandle h = null;
- try {
- h = dbManager.open();
- Iterable<Vertex> q = h.select("SELECT * FROM " + DbVertex.Reservation.pname() +
- " WHERE " + DbConstants.DUCC_DBCAT + "='" + DbCategory.History.pname() + "'" +
- " ORDER BY ducc_dbid DESC LIMIT "+ max);
- for ( Vertex v : q ) {
- IDuccWorkReservation j = restoreReservationInternal(h, (OrientVertex) v);
- ret.add(j);
- }
- } finally {
- h.close();
- }
+ return restoreSeveralThings(IDuccWorkReservation.class, RES_TABLE, max);
+ }
- return ret;
- }
// End of reservations section
// ----------------------------------------------------------------------------------------------------
// ----------------------------------------------------------------------------------------------------
// Services section
- // public void serviceSave(IDuccWorkService s)
- // throws Exception
- // {
- // }
public void saveService(IDuccWorkService s)
throws Exception
{
- saveJobInternal((IDuccWorkJob)s, DbVertex.ServiceInstance, true, DbCategory.History);
- }
-
- public void saveServiceUnsafe(IDuccWorkService s)
- throws Exception
- {
- saveJobInternal((IDuccWorkJob)s, DbVertex.ServiceInstance, false, DbCategory.History);
+ saveWork(servicePrepare, s, true);
}
@@ -590,19 +334,7 @@ public class HistoryManagerDb
public IDuccWorkService restoreService(long duccid)
throws Exception
{
- DbHandle h = null;
- try {
- h = dbManager.open();
- Iterable<Vertex> q = h.select("SELECT * FROM " + DbVertex.ServiceInstance.pname() + " WHERE ducc_dbid=" + duccid +
- " AND " + DbConstants.DUCC_DBCAT + "='" + DbCategory.History.pname() + "'");
- for ( Vertex v : q ) {
- return restoreJobInternal(h, (OrientVertex) v);
- }
- } finally {
- h.close();
- }
-
- return null;
+ return (IDuccWorkService) restoreWork(IDuccWorkService.class, SVC_TABLE, duccid);
}
/**
@@ -611,23 +343,7 @@ public class HistoryManagerDb
public ArrayList<IDuccWorkService> restoreServices(long max)
throws Exception
{
- ArrayList<IDuccWorkService> ret = new ArrayList<IDuccWorkService>();
- DbHandle h = null;
- try {
- h = dbManager.open();
- Iterable<Vertex> q = h.select("SELECT * FROM " + DbVertex.ServiceInstance.pname() +
- " WHERE " + DbConstants.DUCC_DBCAT +"='" + DbCategory.History.pname() + "'" +
- " ORDER BY ducc_dbid DESC LIMIT "+ max);
- for ( Vertex v : q ) {
- IDuccWorkService j = restoreJobInternal(h, (OrientVertex) v);
- ret.add(j);
- }
- } finally {
- h.close();
- }
-
- return ret;
-
+ return restoreSeveralThings(IDuccWorkService.class, SVC_TABLE, max);
}
// End of services section
// ----------------------------------------------------------------------------------------------------
@@ -652,127 +368,92 @@ public class HistoryManagerDb
long now = System.currentTimeMillis();
boolean ret = true;
- DbHandle h = null;
- try {
- h = dbManager.open();
- } catch ( Exception e ) {
- logger.warn(methodName, null, "Cannot open database.", e);
- if ( h != null ) h.close();
- return false;
- }
-
// We transactionally delete the old checkpoint, and then save the new one. If something gows wrong we
// rollback and thus don't lose stuff. In theory.
- try {
- h.execute("DELETE VERTEX V where " + DbConstants.DUCC_DBCAT + "='" + DbCategory.Checkpoint.pname() + "'");
- Map<DuccId, IDuccWork> map = work.getMap();
- for (IDuccWork w : map.values() ) {
- switch(w.getDuccType()) {
- case Job:
- saveJobNoCommit(h, (IDuccWorkJob) w, DbVertex.Job, DbCategory.Checkpoint);
- break;
- case Service:
- saveJobNoCommit(h, (IDuccWorkJob) w, DbVertex.ServiceInstance, DbCategory.Checkpoint);
- break;
- case Reservation:
- if ( w.getDuccId().getFriendly() == 282282 ) {
- int x = 0;
- x++;
- }
- saveReservationNoCommit(h, (IDuccWorkReservation) w, DbCategory.Checkpoint);
- break;
- default:
- break;
- }
- }
-
- Gson g = DbHandle.mkGsonForJob();
- ProcessToJobList l = new ProcessToJobList(processToJob);
- String json = g.toJson(l, l.getClass());
- // logger.info(methodName, null, "ProcessToJob:", json);
- h.saveObject(DbVertex.ProcessToJob, null, json, DbCategory.Checkpoint);
- h.commit();
+
+ // TODO: make the truncate and insert transactional
+ DbHandle h = dbManager.open();
+ h.truncate("ducc.orckpt");
+
+ try {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(baos);
+ out.writeObject(work);
+ out.close();
+ byte[] bytes = baos.toByteArray();
+ ByteBuffer workbuf = ByteBuffer.wrap(bytes);
+
+ baos = new ByteArrayOutputStream();
+ out= new ObjectOutputStream(baos);
+ out.writeObject(processToJob);
+ out.close();
+ bytes = baos.toByteArray();
+ ByteBuffer mapbuf = ByteBuffer.wrap(bytes);
+
+ h = dbManager.open();
+ h.saveObject(ckptPrepare, 0, workbuf, mapbuf);
+
} catch ( Exception e ) {
- if ( h != null ) h.rollback();
logger.error(methodName, null, "Cannot save ProcessToJob map", e);
ret = false;
} finally {
- if ( h != null ) h.close();
if ( ret ) logger.info(methodName, null, "Saved Orchestrator Checkpoint");
}
- logger.info(methodName, null, "Total time to save checkpoint:", System.currentTimeMillis() - now);
- return ret;
+ logger.info(methodName, null, "Total time to save checkpoint:", System.currentTimeMillis() - now);
+ return ret;
}
/**
* Orchestrator checkpoint. Restore the checkpoint from the DB. Caller must initialize
* empty maps, which we fill in.
*/
- public boolean restore(DuccWorkMap work, Map<DuccId, DuccId> processToJob)
+ public Pair<DuccWorkMap, Map<DuccId, DuccId>> restore()
throws Exception
{
- String methodName = "restore";
+ String methodName = "restore";
DbHandle h = null;
- boolean ret = true;
+ Pair<DuccWorkMap, Map<DuccId, DuccId>> ret = new Pair<DuccWorkMap, Map<DuccId, DuccId>>();
try {
h = dbManager.open();
- // Select all the "top level" objects ith DUCC_LIVE=true. When they get restored the
- // attached object will get collected.
-
- Iterable<Vertex> q = h.select("SELECT * FROM V WHERE (" +
- "@CLASS ='" + DbVertex.Job.pname() +
- "' OR " +
- "@CLASS ='" + DbVertex.Reservation.pname() +
- "' OR " +
- "@CLASS ='" + DbVertex.ServiceInstance.pname() +
- "') AND " + DbConstants.DUCC_DBCAT + "='" + DbCategory.Checkpoint.pname() + "'");
-
- IDuccWork w = null;
- for ( Vertex v : q ) {
- String l = ((OrientVertex) v).getLabel();
- if ( l.equals(DbVertex.Job.pname()) || l.equals(DbVertex.ServiceInstance.pname()) ) {
- w = restoreJobInternal(h, (OrientVertex) v);
- } else if ( l.equals(DbVertex.Reservation.pname()) ) {
- w = restoreReservationInternal(h, (OrientVertex) v);
- } else {
- logger.warn(methodName, null, "Unexpected record of type", l, "in checkpoint restore.");
+ String cql = "SELECT * FROM ducc.orckpt WHERE id=0";
+ ResultSet rs = h.execute(cql);
+ for ( Row r : rs ) {
+ logger.info(methodName, null, "Found checkpoint.");
+ ByteBuffer bbWork = r.getBytes("work");
+ ByteBuffer bbmap = r.getBytes("p2jmap");
+
+ byte[] workbytes = bbWork.array();
+ ByteArrayInputStream bais = new ByteArrayInputStream(workbytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ DuccWorkMap work = (DuccWorkMap) ois.readObject();
+ Map<DuccId, IDuccWork> map = work.getMap();
+ ois.close();
+
+ workbytes = bbmap.array();
+ bais = new ByteArrayInputStream(workbytes);
+ ois = new ObjectInputStream(bais);
+ Map<DuccId, DuccId> processToJob = (Map<DuccId, DuccId>) ois.readObject();
+ ois.close();
+
+ // hack because java serializion is stupid and won't call the no-args constructor - need
+ // to restore sometransient fields
+ Set<DuccId> ids = work.getReservationKeySet();
+ for ( DuccId id : ids ) {
+ DuccWorkReservation res = (DuccWorkReservation) work.findDuccWork(DuccType.Reservation, ""+id.getFriendly());
+ if ( r != null ) res.initLogger();
}
- work.addDuccWork(w);
+ ret = new Pair(work, processToJob);
}
- q = h.select("SELECT FROM " + DbVertex.ProcessToJob.pname());
-
- int count = 0;
- for ( Vertex vv : q ) {
- if ( count > 1 ) {
- logger.error(methodName, null, "Oops - we have multiple ProcessToJob records. Using the first one but it may be wrong.");
- break;
- }
-
- OrientVertex v = (OrientVertex) vv;
- ODocument d = v.getRecord();
- String json = d.toJSON();
- logger.info(methodName, null, json);
-
- Gson g = DbHandle.mkGsonForJob();
-
- ProcessToJobList l = g.fromJson(json, ProcessToJobList.class);
- l.fill(processToJob);
- }
-
- } catch ( Exception e ) {
+ } catch ( Exception e ) {
logger.error(methodName, null, "Error restoring checkpoint:", e);
- ret = false;
- } finally {
- if ( h != null ) h.close();
- }
-
-
+ }
+
return ret;
}
-
+
// End of OR checkpoint save and restore
// ----------------------------------------------------------------------------------------------------
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/ReadCkpt.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/ReadCkpt.java?rev=1711088&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/ReadCkpt.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/ReadCkpt.java Wed Oct 28 18:12:53 2015
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+*/
+package org.apache.uima.ducc.database;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.uima.ducc.common.utils.DuccLogger;
+import org.apache.uima.ducc.common.utils.id.DuccId;
+import org.apache.uima.ducc.transport.event.common.DuccWorkMap;
+
+/**
+ * A test routine, read and verify the OR checkpoint file.
+ */
+
+public class ReadCkpt
+{
+
+ DuccLogger logger = DuccLogger.getLogger(DbLoader.class, "DBLOAD");
+ String DUCC_HOME;
+ DbManager dbManager = null;
+ HistoryManagerDb hmd = null;
+
+ public ReadCkpt()
+ throws Exception
+ {
+ DUCC_HOME = System.getProperty("DUCC_HOME");
+ if ( DUCC_HOME == null ) {
+ System.out.println("System proprety -DDUCC_HOME must be set.");
+ System.exit(1);
+ }
+
+ String state_url = "bluej538";
+ System.setProperty("ducc.state.database.url", state_url);
+
+ dbManager = new DbManager(state_url, logger);
+ dbManager.init();
+
+ hmd = new HistoryManagerDb();
+ }
+
+ public void run()
+ {
+ DuccWorkMap work = new DuccWorkMap();
+ Map<DuccId, DuccId> processToJob = new HashMap<DuccId, DuccId>();
+
+ }
+
+ public static void main(String [] args)
+ {
+ try {
+ ReadCkpt rc = new ReadCkpt();
+ rc.run();
+ } catch (Exception e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+
+
+ }
+
+}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/RmStatePersistence.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/RmStatePersistence.java?rev=1711088&r1=1711087&r2=1711088&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/RmStatePersistence.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-database/src/main/java/org/apache/uima/ducc/database/RmStatePersistence.java Wed Oct 28 18:12:53 2015
@@ -19,17 +19,16 @@
package org.apache.uima.ducc.database;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.uima.ducc.common.persistence.rm.IRmPersistence;
import org.apache.uima.ducc.common.utils.DuccLogger;
-import org.apache.uima.ducc.database.DbConstants.DbCategory;
-import org.apache.uima.ducc.database.DbConstants.DbVertex;
-import com.google.gson.Gson;
-import com.tinkerpop.blueprints.impls.orient.OrientVertex;
+import com.datastax.driver.core.SimpleStatement;
/**
* Manage saving and fetching of transient RM state. The primary consumer is
@@ -67,6 +66,11 @@ public class RmStatePersistence
init(stateUrl);
}
+ public void shutdown()
+ {
+ dbManager.shutdown();
+ }
+
public void clear()
throws Exception
{
@@ -74,41 +78,55 @@ public class RmStatePersistence
DbHandle h = null;
try {
h = dbManager.open();
- h.execute("DELETE VERTEX V where " + DbConstants.DUCC_DBCAT + "='" + DbCategory.RmState.pname() + "'");
+ h.execute("TRUNCATE ducc.rmnodes");
} catch ( Exception e ) {
logger.error(methodName, null, "Cannot clear the database.", e);
- } finally {
- if ( h != null ) h.close();
- }
+ }
}
- public String toGson(Object o)
+ static List<SimpleStatement> mkSchema()
+ throws Exception
{
- // We need to define Instance creators and such so we do it in a common place
- Gson g = DbHandle.mkGsonForJob();
- return g.toJson(o);
+ List<SimpleStatement> ret = new ArrayList<SimpleStatement>();
+
+ StringBuffer buf = new StringBuffer("CREATE TABLE IF NOT EXISTS ducc." + RmProperty.TABLE_NAME.pname() + " (");
+ buf.append(DbUtil.mkSchema(RmProperty.values()));
+ buf.append(")");
+ ret.add(new SimpleStatement(buf.toString()));
+ return ret;
}
- public Object createMachine(String m, Properties props)
+ // static String[] mkSchemaItems()
+ // {
+ // int size = RmProperty.values().length;
+ // String[] ret = new String[size];
+ // int ndx = 0;
+
+ // for ( RmProperty n: RmProperty.values() ) {
+ // String s = n.pname();
+ // s = s + " " + DbUtil.typeToString(n.type());
+ // if ( n.isPrimaryKey() ) {
+ // s = s + " PRIMARY KEY";
+ // }
+ // ret[ndx++] = s;
+ // }
+ // return ret;
+ // }
+
+ public void createMachine(String m, Map<RmProperty, Object> props)
throws Exception
{
String methodName = "createMachine";
DbHandle h = dbManager.open();
- Object ret = null;
try {
- OrientVertex v = h.createProperties(DbConstants.DUCC_DBNODE, m, DbVertex.RmNode, DbCategory.RmState, props);
- ret = v.getId();
- h.commit();
+ String cql = DbUtil.mkInsert("ducc.rmnodes", props);
+ h.execute(cql);
} catch ( Exception e ) {
- logger.error(methodName, null, "Update", m, "ROLLBACK: ", e);
- if ( h != null ) h.rollback();
- } finally {
- if ( h != null ) h.close();
- }
- return ret;
+ logger.error(methodName, null, "Error creating new record:", e);
+ }
}
- public void setProperties(Object dbid, String dbk, Object... props)
+ public void setProperties(String node, Object... props)
throws Exception
{
String methodName = "setProperties";
@@ -121,37 +139,28 @@ public class RmStatePersistence
DbHandle h = dbManager.open();
try {
- h.updateProperties(dbid, props);
- h.commit();
+ h.updateProperties("ducc.rmnodes", "WHERE name='" + node + "'", props);
} catch ( Exception e ) {
- logger.error(methodName, null, "Update", dbk, "ROLLBACK: ", e);
- if ( h != null ) h.rollback();
- } finally {
- if ( h != null ) h.close();
- logger.info(methodName, null, "Total time to update properties on", dbid.toString(), System.currentTimeMillis() - now);
+ logger.error(methodName, null, "Problem setting properties");
+ } finally {
+ logger.info(methodName, null, "Total time to update properties on", System.currentTimeMillis() - now);
}
}
- public void setProperty(Object dbid, String dbk, RmPropName k, Object v)
+ public void setProperty(String node, RmProperty k, Object v)
throws Exception
{
String methodName = "setProperty";
- long now = System.currentTimeMillis();
DbHandle h = dbManager.open();
try {
- h.updateProperty(dbid, k.pname(), v);
- h.commit();
+ h.updateProperty("ducc.rmnodes", "name='" + node + "'", k.columnName(), v);
} catch ( Exception e ) {
- logger.error(methodName, null, "Update", dbk, "ROLLBACK: ", e);
- if ( h != null ) h.rollback();
- } finally {
- if ( h != null ) h.close();
- logger.info(methodName, null, "Total time to update property on", dbid.toString(), System.currentTimeMillis() - now);
- }
+ logger.error(methodName, null, "Problem setting properties.");
+ }
}
@@ -172,28 +181,3 @@ public class RmStatePersistence
}
}
-
-/**
- String name;
- String nodepoolId;
- long memory;
- int order;
- boolean blacklisted; // UIMA-4142
- boolean online; // UIMA-4234
- boolean responsive; // UIMA-4234
-
-
-
-
- Properties file for a node
- name = string
- ip = string
- state = <state>
- states: vary status: online | offline
- reporting : present | absent
- nodepool = string
- quantum = string
- class = string
- scheduling policy = string
- scheduled work = list of duccids of work on the node
- */