/*
* Copyright 2012 LinkedIn Corp.
*
* 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.
*/
package azkaban.trigger;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.ResultSetHandler;
import org.apache.log4j.Logger;
import org.joda.time.DateTime;
import azkaban.database.AbstractJdbcLoader;
import azkaban.utils.GZIPUtils;
import azkaban.utils.JSONUtils;
import azkaban.utils.Props;
public class JdbcTriggerLoader extends AbstractJdbcLoader implements TriggerLoader {
private static Logger logger = Logger.getLogger(JdbcTriggerLoader.class);
private EncodingType defaultEncodingType = EncodingType.GZIP;
private static final String triggerTblName = "triggers";
private static final String GET_UPDATED_TRIGGERS =
"SELECT trigger_id, trigger_source, modify_time, enc_type, data FROM " + triggerTblName + " WHERE modify_time>=?";
private static String GET_ALL_TRIGGERS =
"SELECT trigger_id, trigger_source, modify_time, enc_type, data FROM " + triggerTblName;
private static String GET_TRIGGER =
"SELECT trigger_id, trigger_source, modify_time, enc_type, data FROM " + triggerTblName + " WHERE trigger_id=?";
private static String ADD_TRIGGER =
"INSERT INTO " + triggerTblName + " ( modify_time) values (?)";
private static String REMOVE_TRIGGER =
"DELETE FROM " + triggerTblName + " WHERE trigger_id=?";
private static String UPDATE_TRIGGER =
"UPDATE " + triggerTblName + " SET trigger_source=?, modify_time=?, enc_type=?, data=? WHERE trigger_id=?";
public EncodingType getDefaultEncodingType() {
return defaultEncodingType;
}
public void setDefaultEncodingType(EncodingType defaultEncodingType) {
this.defaultEncodingType = defaultEncodingType;
}
public JdbcTriggerLoader(Props props) {
super(props);
}
@Override
public List<Trigger> getUpdatedTriggers(long lastUpdateTime) throws TriggerLoaderException {
logger.info("Loading triggers changed since " + new DateTime(lastUpdateTime).toString());
Connection connection = getConnection();
QueryRunner runner = new QueryRunner();
ResultSetHandler<List<Trigger>> handler = new TriggerResultHandler();
List<Trigger> triggers;
try {
triggers = runner.query(connection, GET_UPDATED_TRIGGERS, handler, lastUpdateTime);
} catch (SQLException e) {
logger.error(GET_ALL_TRIGGERS + " failed.");
throw new TriggerLoaderException("Loading triggers from db failed. ", e);
} finally {
DbUtils.closeQuietly(connection);
}
logger.info("Loaded " + triggers.size() + " triggers.");
return triggers;
}
@Override
public List<Trigger> loadTriggers() throws TriggerLoaderException {
logger.info("Loading all triggers from db.");
Connection connection = getConnection();
QueryRunner runner = new QueryRunner();
ResultSetHandler<List<Trigger>> handler = new TriggerResultHandler();
List<Trigger> triggers;
try {
triggers = runner.query(connection, GET_ALL_TRIGGERS, handler);
} catch (SQLException e) {
logger.error(GET_ALL_TRIGGERS + " failed.");
throw new TriggerLoaderException("Loading triggers from db failed. ", e);
} finally {
DbUtils.closeQuietly(connection);
}
logger.info("Loaded " + triggers.size() + " triggers.");
return triggers;
}
@Override
public void removeTrigger(Trigger t) throws TriggerLoaderException {
logger.info("Removing trigger " + t.toString() + " from db.");
QueryRunner runner = createQueryRunner();
try {
int removes = runner.update(REMOVE_TRIGGER, t.getTriggerId());
if (removes == 0) {
throw new TriggerLoaderException("No trigger has been removed.");
}
} catch (SQLException e) {
logger.error(REMOVE_TRIGGER + " failed.");
throw new TriggerLoaderException("Remove trigger " + t.toString() + " from db failed. ", e);
}
}
@Override
public void addTrigger(Trigger t) throws TriggerLoaderException {
logger.info("Inserting trigger " + t.toString() + " into db.");
t.setLastModifyTime(System.currentTimeMillis());
Connection connection = getConnection();
try {
addTrigger(connection, t, defaultEncodingType);
}
catch (Exception e) {
throw new TriggerLoaderException("Error uploading trigger", e);
}
finally {
DbUtils.closeQuietly(connection);
}
}
private synchronized void addTrigger(Connection connection, Trigger t, EncodingType encType) throws TriggerLoaderException {
QueryRunner runner = new QueryRunner();
long id;
try {
runner.update(connection, ADD_TRIGGER, DateTime.now().getMillis());
connection.commit();
id = runner.query(connection, LastInsertID.LAST_INSERT_ID, new LastInsertID());
if (id == -1l) {
logger.error("trigger id is not properly created.");
throw new TriggerLoaderException("trigger id is not properly created.");
}
t.setTriggerId((int)id);
updateTrigger(t);
logger.info("uploaded trigger " + t.getDescription());
} catch (SQLException e) {
throw new TriggerLoaderException("Error creating trigger.", e);
}
}
@Override
public void updateTrigger(Trigger t) throws TriggerLoaderException {
logger.info("Updating trigger " + t.getTriggerId() + " into db.");
t.setLastModifyTime(System.currentTimeMillis());
Connection connection = getConnection();
try{
updateTrigger(connection, t, defaultEncodingType);
}
catch(Exception e) {
e.printStackTrace();
throw new TriggerLoaderException("Failed to update trigger " + t.toString() + " into db!");
}
finally {
DbUtils.closeQuietly(connection);
}
}
private void updateTrigger(Connection connection, Trigger t, EncodingType encType) throws TriggerLoaderException {
String json = JSONUtils.toJSON(t.toJson());
byte[] data = null;
try {
byte[] stringData = json.getBytes("UTF-8");
data = stringData;
if (encType == EncodingType.GZIP) {
data = GZIPUtils.gzipBytes(stringData);
}
logger.debug("NumChars: " + json.length() + " UTF-8:" + stringData.length + " Gzip:"+ data.length);
}
catch (IOException e) {
throw new TriggerLoaderException("Error encoding the trigger " + t.toString());
}
QueryRunner runner = new QueryRunner();
try {
int updates = runner.update( connection,
UPDATE_TRIGGER,
t.getSource(),
t.getLastModifyTime(),
encType.getNumVal(),
data,
t.getTriggerId());
connection.commit();
if (updates == 0) {
throw new TriggerLoaderException("No trigger has been updated.");
//logger.error("No trigger is updated!");
} else {
logger.info("Updated " + updates + " records.");
}
} catch (SQLException e) {
logger.error(UPDATE_TRIGGER + " failed.");
throw new TriggerLoaderException("Update trigger " + t.toString() + " into db failed. ", e);
}
}
private static class LastInsertID implements ResultSetHandler<Long> {
private static String LAST_INSERT_ID = "SELECT LAST_INSERT_ID()";
@Override
public Long handle(ResultSet rs) throws SQLException {
if (!rs.next()) {
return -1l;
}
long id = rs.getLong(1);
return id;
}
}
public class TriggerResultHandler implements ResultSetHandler<List<Trigger>> {
@Override
public List<Trigger> handle(ResultSet rs) throws SQLException {
if (!rs.next()) {
return Collections.<Trigger>emptyList();
}
ArrayList<Trigger> triggers = new ArrayList<Trigger>();
do {
int triggerId = rs.getInt(1);
// String triggerSource = rs.getString(2);
// long modifyTime = rs.getLong(3);
int encodingType = rs.getInt(4);
byte[] data = rs.getBytes(5);
Object jsonObj = null;
if (data != null) {
EncodingType encType = EncodingType.fromInteger(encodingType);
try {
// Convoluted way to inflate strings. Should find common package or helper function.
if (encType == EncodingType.GZIP) {
// Decompress the sucker.
String jsonString = GZIPUtils.unGzipString(data, "UTF-8");
jsonObj = JSONUtils.parseJSONFromString(jsonString);
}
else {
String jsonString = new String(data, "UTF-8");
jsonObj = JSONUtils.parseJSONFromString(jsonString);
}
} catch (IOException e) {
throw new SQLException("Error reconstructing trigger data " );
}
}
Trigger t = null;
try {
t = Trigger.fromJson(jsonObj);
triggers.add(t);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
logger.error("Failed to load trigger " + triggerId);
}
} while (rs.next());
return triggers;
}
}
private Connection getConnection() throws TriggerLoaderException {
Connection connection = null;
try {
connection = super.getDBConnection(false);
} catch (Exception e) {
DbUtils.closeQuietly(connection);
throw new TriggerLoaderException("Error getting DB connection.", e);
}
return connection;
}
@Override
public Trigger loadTrigger(int triggerId) throws TriggerLoaderException {
logger.info("Loading trigger " + triggerId + " from db.");
Connection connection = getConnection();
QueryRunner runner = new QueryRunner();
ResultSetHandler<List<Trigger>> handler = new TriggerResultHandler();
List<Trigger> triggers;
try {
triggers = runner.query(connection, GET_TRIGGER, handler, triggerId);
} catch (SQLException e) {
logger.error(GET_TRIGGER + " failed.");
throw new TriggerLoaderException("Loading trigger from db failed. ", e);
} finally {
DbUtils.closeQuietly(connection);
}
if(triggers.size() == 0) {
logger.error("Loaded 0 triggers. Failed to load trigger " + triggerId);
throw new TriggerLoaderException("Loaded 0 triggers. Failed to load trigger " + triggerId);
}
return triggers.get(0);
}
}