innov
/
aggregation-platform
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

行政区划字母大写--校验时启动kuber失败的问题

Browse Source
web_backend_develope
chenlw 8 years ago
parent e58679d4a7
commit 6161d131e6
4 changed files with 960 additions and 959 deletions

410
src/com/platform/service/OracleStatusService.java
Unescape View File

@ -1,206 +1,206 @@
package com.platform.service; package com.platform.service;
import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.ReplicationController; import io.fabric8.kubernetes.api.model.ReplicationController;
import java.util.Hashtable; import java.util.Hashtable;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Timer; import java.util.Timer;
import java.util.TimerTask; import java.util.TimerTask;
import com.platform.entities.OracleConnectorParams; import com.platform.entities.OracleConnectorParams;
import com.platform.kubernetes.SimpleKubeClient; import com.platform.kubernetes.SimpleKubeClient;
import com.platform.oracle.OracleConnector; import com.platform.oracle.OracleConnector;
import com.platform.utils.Configs; import com.platform.utils.Configs;
import com.platform.utils.Constant; import com.platform.utils.Constant;
/** /**
* 计时 尝试连接oracle的时间，（超过10分钟判断失败） * 计时 尝试连接oracle的时间，（超过10分钟判断失败）
* *
* @author chen * @author chen
* *
*/ */
public class OracleStatusService { public class OracleStatusService {
private static Map<String, Timer> alliveTask = new Hashtable<String, Timer>(); private static Map<String, Timer> alliveTask = new Hashtable<String, Timer>();
public final static int EXEC_TIME = 9;// 连接多少次后不成功，取消链接 public final static int EXEC_TIME = 9;// 连接多少次后不成功，取消链接
public final static long INTERVAL_TIME = 60 * 1000;// 每隔多少毫秒执行一次连接任务 public final static long INTERVAL_TIME = 60 * 1000;// 每隔多少毫秒执行一次连接任务
public final static long DELAY_TIME = 30 * 1000; // 延迟多少秒后执行 public final static long DELAY_TIME = 30 * 1000; // 延迟多少秒后执行
public void connectToOracle(String replicasName) { public void connectToOracle(String replicasName) {
SimpleKubeClient sKubeClient = new SimpleKubeClient(); SimpleKubeClient sKubeClient = new SimpleKubeClient();
if (alliveTask.containsKey(replicasName)) { if (alliveTask.containsKey(replicasName)) {
killAlliveTask(replicasName); killAlliveTask(replicasName);
} }
OracleConnectorParams orp = new OracleConnectorParams(); OracleConnectorParams orp = new OracleConnectorParams();
Timer timer = new Timer(); Timer timer = new Timer();
alliveTask.put(replicasName, timer); alliveTask.put(replicasName, timer);
timer.schedule(new connectTask(replicasName, orp, sKubeClient), timer.schedule(new connectTask(replicasName, orp, sKubeClient),
DELAY_TIME, INTERVAL_TIME); DELAY_TIME, INTERVAL_TIME);
} }
public void cancelToOracle(String replicasName, String operate) { public void cancelToOracle(String replicasName, String operate) {
if (operate.equals("stop")) { if (operate.equals("stop")) {
String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + replicasName + " status=0"; + " label --overwrite rc " + replicasName + " status=0";
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String str : rList) for (String str : rList)
sb.append(str).append("\n"); sb.append(str).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString()); Configs.CONSOLE_LOGGER.info(sb.toString());
// SimpleKubeClient sKubeClient = new SimpleKubeClient(); // SimpleKubeClient sKubeClient = new SimpleKubeClient();
// sKubeClient.updateOrAddReplicasLabelById(replicasName, // sKubeClient.updateOrAddReplicasLabelById(replicasName,
// "status","0"); // "status","0");
} }
killAlliveTask(replicasName); killAlliveTask(replicasName);
} }
/** /**
* 取消并移除指定定时任务 * 取消并移除指定定时任务
* *
* *
* @param taskName * @param taskName
*/ */
public void killAlliveTask(String taskName) { public void killAlliveTask(String taskName) {
if (alliveTask.containsKey(taskName)) { if (alliveTask.containsKey(taskName)) {
alliveTask.get(taskName).cancel(); alliveTask.get(taskName).cancel();
alliveTask.remove(taskName); alliveTask.remove(taskName);
} }
} }
public void killAlliveTasks(String... tasksName) { public void killAlliveTasks(String... tasksName) {
for (String taskName : tasksName) for (String taskName : tasksName)
killAlliveTask(taskName); killAlliveTask(taskName);
} }
/** /**
* 清空定时任务 * 清空定时任务
*/ */
public void cleanUpAlliveTask() { public void cleanUpAlliveTask() {
Iterator<Map.Entry<String, Timer>> iterator = alliveTask.entrySet() Iterator<Map.Entry<String, Timer>> iterator = alliveTask.entrySet()
.iterator(); .iterator();
while (iterator.hasNext()) { while (iterator.hasNext()) {
Map.Entry<String, Timer> entry = iterator.next(); Map.Entry<String, Timer> entry = iterator.next();
entry.getValue().cancel(); entry.getValue().cancel();
} }
alliveTask.clear(); alliveTask.clear();
} }
/** /**
* 链接oracle任务类 * 链接oracle任务类
* *
* @author wuming * @author wuming
* *
*/ */
class connectTask extends TimerTask { class connectTask extends TimerTask {
private String taskName; private String taskName;
private int count; private int count;
private OracleConnectorParams ocp; private OracleConnectorParams ocp;
private SimpleKubeClient client; private SimpleKubeClient client;
public connectTask(String taskName, OracleConnectorParams ocp, public connectTask(String taskName, OracleConnectorParams ocp,
SimpleKubeClient client) { SimpleKubeClient client) {
this.taskName = taskName; this.taskName = taskName;
this.ocp = ocp; this.ocp = ocp;
this.count = 0; this.count = 0;
this.client = client; this.client = client;
} }
public connectTask(OracleConnectorParams ocp, SimpleKubeClient client) { public connectTask(OracleConnectorParams ocp, SimpleKubeClient client) {
this.taskName = ocp.getName(); this.taskName = ocp.getName();
this.ocp = ocp; this.ocp = ocp;
this.count = 0; this.count = 0;
this.client = client; this.client = client;
} }
@Override @Override
public void run() { public void run() {
if (count == EXEC_TIME && alliveTask.containsKey(taskName)) { // 如果任务已经执行10次，则任务oracle启动失败，并取消oracle连接 if (count == EXEC_TIME && alliveTask.containsKey(taskName)) { // 如果任务已经执行10次，则任务oracle启动失败，并取消oracle连接
killAlliveTask(taskName); killAlliveTask(taskName);
// client.updateOrAddReplicasLabelById(taskName, "status", "1"); // client.updateOrAddReplicasLabelById(taskName, "status", "1");
// //更新ReplicationController标签，将oracle状态标示未1(0:启动中，1：失败，2：成功) // //更新ReplicationController标签，将oracle状态标示未1(0:启动中，1：失败，2：成功)
String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + taskName + " status=1"; + " label --overwrite rc " + taskName + " status=1";
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String str : rList) for (String str : rList)
sb.append(str).append("\n"); sb.append(str).append("\n");
Configs.CONSOLE_LOGGER.info("更新replicationController标签： " Configs.CONSOLE_LOGGER.info("更新replicationController标签： "
+ taskName + "\t[标签更新为： 失败]"); + taskName + "\t[标签更新为： 失败]");
} else { // 否则，执行连接oracle任务，判断oracle是否启动成功 } else { // 否则，执行连接oracle任务，判断oracle是否启动成功
Pod pod = filterPod(); Pod pod = filterPod();
if (pod != null) { if (pod != null) {
String ip = client.getPodHostIp(pod); String ip = client.getPodHostIp(pod);
int port = client.getPodContainerport(pod); int port = client.getPodContainerport(pod);
if (ip != null && port != 0) { if (ip != null && port != 0) {
String url = "jdbc:oracle:thin:@" + ip + ":" + port String url = "jdbc:oracle:thin:@" + ip + ":" + port
+ ":" + ocp.getDatabaseName(); + ":" + ocp.getDatabaseName();
boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败 boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败
ocp.getUser(), ocp.getPassword()); ocp.getUser(), ocp.getPassword());
Configs.CONSOLE_LOGGER.info("url:" + url + ",user:" Configs.CONSOLE_LOGGER.info("url:" + url + ",user:"
+ ocp.getUser() + ",password:" + ocp.getUser() + ",password:"
+ ocp.getPassword()); + ocp.getPassword());
String message = "失败"; String message = "失败";
if (flag && alliveTask.containsKey(taskName)) { if (flag && alliveTask.containsKey(taskName)) {
String cmd = "kubectl --server " String cmd = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + taskName + " label --overwrite rc " + taskName
+ " status=2"; + " status=2";
// client.updateOrAddReplicasLabelById(taskNSyame, // client.updateOrAddReplicasLabelById(taskNSyame,
// "status", "2"); // "status", "2");
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String str : rList) for (String str : rList)
sb.append(str).append("\n"); sb.append(str).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString()); Configs.CONSOLE_LOGGER.info(sb.toString());
message = "成功"; message = "成功";
killAlliveTask(taskName); // 连接成功，取消连接 killAlliveTask(taskName); // 连接成功，取消连接
Configs.CONSOLE_LOGGER Configs.CONSOLE_LOGGER
.info("更新replicationController标签： " .info("更新replicationController标签： "
+ taskName + "\t[标签更新为： 成功]"); + taskName + "\t[标签更新为： 成功]");
} }
Configs.CONSOLE_LOGGER.info("连接到数据库服务： " + taskName Configs.CONSOLE_LOGGER.info("连接到数据库服务： " + taskName
+ "\t[连接结果： " + message + "]"); + "\t[连接结果： " + message + "]");
} }
} }
} }
count++; count++;
} }
/** /**
* 获取oracle的连接ip地址和端口号 * 获取oracle的连接ip地址和端口号
* *
* @return * @return
*/ */
private Pod filterPod() { private Pod filterPod() {
Pod pod = null; Pod pod = null;
ReplicationController replicationController = client ReplicationController replicationController = client
.getReplicationController(taskName); .getReplicationController(taskName);
if (null != replicationController) { if (null != replicationController) {
List<Pod> filterPods = client List<Pod> filterPods = client
.getPodsForApplicaList(replicationController); .getPodsForApplicaList(replicationController);
if (filterPods != null && filterPods.size() > 0) { if (filterPods != null && filterPods.size() > 0) {
pod = filterPods.get(0); pod = filterPods.get(0);
} }
} }
return pod; return pod;
} }
public String getTaskName() { public String getTaskName() {
return taskName; return taskName;
} }
public void setTaskName(String taskName) { public void setTaskName(String taskName) {
this.taskName = taskName; this.taskName = taskName;
} }
public int getCount() { public int getCount() {
return count; return count;
} }
} }
} }

1
src/com/platform/service/impl/CheckoutServiceImpl.java
Unescape View File

@ -200,6 +200,7 @@ public class CheckoutServiceImpl implements ICheckoutService {
int listSize = resultList.size(); int listSize = resultList.size();
for (int i = 0; i < listSize; i++) { for (int i = 0; i < listSize; i++) {
CheckoutEntity obj = resultList.get(i); CheckoutEntity obj = resultList.get(i);
obj.setAreaCode(obj.getAreaCode().toLowerCase());
if (null != obj && "y".equals(obj.getCollUpdate())) { if (null != obj && "y".equals(obj.getCollUpdate())) {
obj.setCollUpdate("是"); obj.setCollUpdate("是");
} }

838
src/com/platform/service/impl/OracleExtractServiceImpl.java
Unescape View File

@ -1,419 +1,419 @@
package com.platform.service.impl; package com.platform.service.impl;
import java.sql.Connection; import java.sql.Connection;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
import javax.annotation.Resource; import javax.annotation.Resource;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import com.base.Custom4exception; import com.base.Custom4exception;
import com.base.CustomException; import com.base.CustomException;
import com.platform.dao.DataInfoDao; import com.platform.dao.DataInfoDao;
import com.platform.entities.DataInfoEntity; import com.platform.entities.DataInfoEntity;
import com.platform.entities.GatherOracleInfo; import com.platform.entities.GatherOracleInfo;
import com.platform.entities.OracleConnectorParams; import com.platform.entities.OracleConnectorParams;
import com.platform.kubernetes.SimpleKubeClient; import com.platform.kubernetes.SimpleKubeClient;
import com.platform.oracle.OracleConnector; import com.platform.oracle.OracleConnector;
import com.platform.service.IOracleExtractService; import com.platform.service.IOracleExtractService;
import com.platform.service.OracleExtractHelper; import com.platform.service.OracleExtractHelper;
import com.platform.utils.CacheSetCantDelete; import com.platform.utils.CacheSetCantDelete;
import com.platform.utils.Configs; import com.platform.utils.Configs;
import com.platform.utils.Constant; import com.platform.utils.Constant;
import com.platform.utils.DateForm; import com.platform.utils.DateForm;
import com.platform.utils.FileOperateHelper; import com.platform.utils.FileOperateHelper;
@Service(value = "OracleExtract") @Service(value = "OracleExtract")
public class OracleExtractServiceImpl implements IOracleExtractService { public class OracleExtractServiceImpl implements IOracleExtractService {
/** /**
* 日志 * 日志
*/ */
public final static Logger log = Logger public final static Logger log = Logger
.getLogger(OracleExtractServiceImpl.class); .getLogger(OracleExtractServiceImpl.class);
@Resource(name = "dataInfoDao") @Resource(name = "dataInfoDao")
private DataInfoDao dataInfoDao; private DataInfoDao dataInfoDao;
/** /**
* kubernetes client * kubernetes client
*/ */
private SimpleKubeClient client = new SimpleKubeClient(); private SimpleKubeClient client = new SimpleKubeClient();
/** /**
* 抽取 * 抽取
*/ */
private OracleExtractHelper oracleExtract = new OracleExtractHelper(); private OracleExtractHelper oracleExtract = new OracleExtractHelper();
/** /**
* 数据库连接实现类 * 数据库连接实现类
*/ */
private OracleConnector connect = new OracleConnector(); private OracleConnector connect = new OracleConnector();
@Override @Override
public boolean extractOracle(String name, public boolean extractOracle(String name,
List<OracleConnectorParams> datainfos, GatherOracleInfo oracleModel) List<OracleConnectorParams> datainfos, GatherOracleInfo oracleModel)
throws Exception { throws Exception {
boolean isSuccess = false; boolean isSuccess = false;
try { try {
// map转 bean(汇总库信息-带tableName的) // map转 bean(汇总库信息-带tableName的)
// GatherOracleInfo oracleModel = oracleConnect; // GatherOracleInfo oracleModel = oracleConnect;
// 采集库连接参数 // 采集库连接参数
// List<OracleConnectorParams> datainfos = dataInfolist; // List<OracleConnectorParams> datainfos = dataInfolist;
if (datainfos.size() == 0) { if (datainfos.size() == 0) {
return false; return false;
} }
Connection conn = OracleConnector.connectionBuilder( Connection conn = OracleConnector.connectionBuilder(
"jdbc:oracle:thin:@" + oracleModel.getIp() + ":" "jdbc:oracle:thin:@" + oracleModel.getIp() + ":"
+ oracleModel.getPort() + ":" + oracleModel.getPort() + ":"
+ oracleModel.getDatabaseName(), + oracleModel.getDatabaseName(),
oracleModel.getUser(), oracleModel.getPassword(), oracleModel.getUser(), oracleModel.getPassword(),
datainfos.get(0)); datainfos.get(0));
if (null == conn) { if (null == conn) {
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
+ datainfos.get(0).getName() + ".log", + datainfos.get(0).getName() + ".log",
"创建oracle连接失败: [" + conn + "]\r\n"); "创建oracle连接失败: [" + conn + "]\r\n");
return false; return false;
} }
for (OracleConnectorParams collectOracle : datainfos) { for (OracleConnectorParams collectOracle : datainfos) {
String replicasName = collectOracle.getName(); String replicasName = collectOracle.getName();
DataInfoEntity data = new DataInfoEntity(); DataInfoEntity data = new DataInfoEntity();
try { try {
if (null != collectOracle.getDataId() if (null != collectOracle.getDataId()
&& !"".equals(collectOracle.getDataId())) { && !"".equals(collectOracle.getDataId())) {
data.setId(Integer.valueOf(collectOracle.getDataId())); data.setId(Integer.valueOf(collectOracle.getDataId()));
data.setExtractStatus(1); data.setExtractStatus(1);
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
collectOracle.setName("J" collectOracle.setName("J"
+ collectOracle.getName().replace("-", "_")); + collectOracle.getName().replace("-", "_"));
String cmd = "kubectl --server " String cmd = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + replicasName + " label --overwrite rc " + replicasName
+ " isExtract=1"; + " isExtract=1";
// sql日志记录时间： // sql日志记录时间：
FileOperateHelper FileOperateHelper
.fileWrite( .fileWrite(
Configs.EXTRACT_LOG_LOCALTION Configs.EXTRACT_LOG_LOCALTION
+ collectOracle.getName() + collectOracle.getName()
+ ".log", + ".log",
"\r\n 开始汇总 \r\n" "\r\n 开始汇总 \r\n"
+ DateForm + DateForm
.date2StringBysecond(new Date()) .date2StringBysecond(new Date())
+ "\r\n"); + "\r\n");
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String string : rList) for (String string : rList)
sb.append(string).append("\n"); sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString()); Configs.CONSOLE_LOGGER.info(sb.toString());
// client.updateOrAddReplicasLabelById(collectOracle.getName(), // client.updateOrAddReplicasLabelById(collectOracle.getName(),
// "isExtract", "1"); // "isExtract", "1");
// //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
oracleExtract.createDBLink(conn, collectOracle); // 创建dblink oracleExtract.createDBLink(conn, collectOracle); // 创建dblink
oracleExtract.createTableSpace(conn, collectOracle, oracleExtract.createTableSpace(conn, collectOracle,
oracleModel); // 创建表空间 oracleModel); // 创建表空间
oracleExtract.createUser(conn, collectOracle, oracleExtract.createUser(conn, collectOracle,
oracleModel);// 创建用户并授权 oracleModel);// 创建用户并授权
oracleExtract.extractColleDB(conn, collectOracle, oracleExtract.extractColleDB(conn, collectOracle,
oracleModel);// 执行抽取 oracleModel);// 执行抽取
// client.updateOrAddReplicasLabelById(collectOracle.getName(), // client.updateOrAddReplicasLabelById(collectOracle.getName(),
// "isExtract", "2"); // "isExtract", "2");
// //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + replicasName + " label --overwrite rc " + replicasName
+ " isExtract=2"; + " isExtract=2";
rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
sb = new StringBuffer(); sb = new StringBuffer();
for (String string : rList) for (String string : rList)
sb.append(string).append("\n"); sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString()); Configs.CONSOLE_LOGGER.info(sb.toString());
data.setExtractStatus(2); data.setExtractStatus(2);
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
DataInfoEntity tmpdata = dataInfoDao.findById(data DataInfoEntity tmpdata = dataInfoDao.findById(data
.getId()); .getId());
data.setId(tmpdata.getSrcId()); data.setId(tmpdata.getSrcId());
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
} }
} catch (Exception e) { } catch (Exception e) {
String cmd = "kubectl label --overwrite rc " + replicasName String cmd = "kubectl label --overwrite rc " + replicasName
+ " isExtract=0"; + " isExtract=0";
Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
data.setExtractStatus(0); data.setExtractStatus(0);
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
DataInfoEntity tmpdata = dataInfoDao.findById(data.getId()); DataInfoEntity tmpdata = dataInfoDao.findById(data.getId());
data.setId(tmpdata.getSrcId()); data.setId(tmpdata.getSrcId());
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
log.error(Custom4exception.OracleSQL_Except, e); log.error(Custom4exception.OracleSQL_Except, e);
} finally { } finally {
// 去掉保存的当前数据id, // 去掉保存的当前数据id,
CacheSetCantDelete.removeExtractId(collectOracle CacheSetCantDelete.removeExtractId(collectOracle
.getDataId()); .getDataId());
String msg = "汇总结束"; String msg = "汇总结束";
if (2 != data.getExtractStatus()) { if (2 != data.getExtractStatus()) {
msg += " 汇总有异常！状态重置为待汇总 "; msg += " 汇总有异常！状态重置为待汇总 ";
data.setExtractStatus(0); data.setExtractStatus(0);
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
DataInfoEntity tmpdata = dataInfoDao.findById(data DataInfoEntity tmpdata = dataInfoDao.findById(data
.getId()); .getId());
data.setId(tmpdata.getSrcId()); data.setId(tmpdata.getSrcId());
dataInfoDao.updateExtract(data); dataInfoDao.updateExtract(data);
} }
// sql日志记录时间： // sql日志记录时间：
FileOperateHelper.fileWrite( FileOperateHelper.fileWrite(
Configs.EXTRACT_LOG_LOCALTION Configs.EXTRACT_LOG_LOCALTION
+ collectOracle.getName() + ".log", + collectOracle.getName() + ".log",
"\r\n " + msg + " >>>>>>> " "\r\n " + msg + " >>>>>>> "
+ DateForm.date2StringBysecond(new Date()) + DateForm.date2StringBysecond(new Date())
+ "\r\n\r\n\n"); + "\r\n\r\n\n");
} }
} }
isSuccess = true; isSuccess = true;
} catch (Exception e) { } catch (Exception e) {
new CustomException(Custom4exception.OracleSQL_Except, e); new CustomException(Custom4exception.OracleSQL_Except, e);
} }
return isSuccess; return isSuccess;
} }
@Override @Override
public boolean isConnectTotalOracle(GatherOracleInfo oracleModel) public boolean isConnectTotalOracle(GatherOracleInfo oracleModel)
throws Exception { throws Exception {
boolean isConnect = false; boolean isConnect = false;
Connection conn = OracleConnector.connectionBuilder( Connection conn = OracleConnector.connectionBuilder(
"jdbc:oracle:thin:@" + oracleModel.getIp() + ":" "jdbc:oracle:thin:@" + oracleModel.getIp() + ":"
+ oracleModel.getPort() + ":" + oracleModel.getPort() + ":"
+ oracleModel.getDatabaseName(), oracleModel.getUser(), + oracleModel.getDatabaseName(), oracleModel.getUser(),
oracleModel.getPassword(), null); oracleModel.getPassword(), null);
if (null == conn) { if (null == conn) {
isConnect = false; isConnect = false;
throw new CustomException(Custom4exception.connect_Oracle_Except, throw new CustomException(Custom4exception.connect_Oracle_Except,
null, oracleModel); null, oracleModel);
// FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION // FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
// + dataInfolist.get(0).getName(), "创建oracle连接失败: [" + conn + // + dataInfolist.get(0).getName(), "创建oracle连接失败: [" + conn +
// "]\r\n"); // "]\r\n");
} else { } else {
isConnect = oracleExtract.testConnect(conn); isConnect = oracleExtract.testConnect(conn);
conn.close(); // 关闭连接 conn.close(); // 关闭连接
} }
return isConnect; return isConnect;
} }
@Override @Override
public boolean extractStandardTable(String name, public boolean extractStandardTable(String name,
List<OracleConnectorParams> dataInfolist, List<OracleConnectorParams> dataInfolist,
GatherOracleInfo oracleConnect) throws Exception { GatherOracleInfo oracleConnect) throws Exception {
boolean isSuccess = false; boolean isSuccess = false;
// map转 bean(汇总库信息-带tableName的) // map转 bean(汇总库信息-带tableName的)
GatherOracleInfo oracleModel = oracleConnect; GatherOracleInfo oracleModel = oracleConnect;
// 采集库连接参数 // 采集库连接参数
List<OracleConnectorParams> datainfos = dataInfolist; List<OracleConnectorParams> datainfos = dataInfolist;
if (datainfos.size() == 0) { if (datainfos.size() == 0) {
return false; return false;
} }
Connection conn = OracleConnector.connectionBuilder( Connection conn = OracleConnector.connectionBuilder(
"jdbc:oracle:thin:@" + oracleModel.getIp() + ":" "jdbc:oracle:thin:@" + oracleModel.getIp() + ":"
+ oracleModel.getPort() + ":" + oracleModel.getPort() + ":"
+ oracleModel.getDatabaseName(), oracleModel.getUser(), + oracleModel.getDatabaseName(), oracleModel.getUser(),
oracleModel.getPassword(), dataInfolist.get(0)); oracleModel.getPassword(), dataInfolist.get(0));
if (null == conn) { if (null == conn) {
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
+ dataInfolist.get(0).getName() + ".log", "创建oracle连接失败: [" + dataInfolist.get(0).getName() + ".log", "创建oracle连接失败: ["
+ conn + "]\r\n"); + conn + "]\r\n");
return false; return false;
} }
for (OracleConnectorParams collectOracle : datainfos) { for (OracleConnectorParams collectOracle : datainfos) {
String replicasName = collectOracle.getName(); String replicasName = collectOracle.getName();
DataInfoEntity data = new DataInfoEntity(); DataInfoEntity data = new DataInfoEntity();
try { try {
if (null != collectOracle.getDataId() if (null != collectOracle.getDataId()
&& !"".equals(collectOracle.getDataId())) { && !"".equals(collectOracle.getDataId())) {
data.setId(Integer.valueOf(collectOracle.getDataId())); data.setId(Integer.valueOf(collectOracle.getDataId()));
// 设置为 标准表 抽取中 // 设置为 标准表 抽取中
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX); data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX);
data.setStandardExtractStatus("1"); data.setStandardExtractStatus("1");
dataInfoDao.update(data); dataInfoDao.update(data);
collectOracle.setName("CQ" collectOracle.setName("CQ"
+ collectOracle.getName().replace("-", "_")); + collectOracle.getName().replace("-", "_"));
String cmd = "kubectl --server " String cmd = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + replicasName + " annotate --overwrite rc " + replicasName
+ " standardExtractStatus=1"; + " standardExtractStatus=1";
// sql日志记录时间： // sql日志记录时间：
FileOperateHelper.fileWrite( FileOperateHelper.fileWrite(
Configs.EXTRACT_STANDARD_LOG_LOCALTION Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ collectOracle.getName() + ".log", + collectOracle.getName() + ".log",
"\r\n 开始抽取标准表 \r\n " "\r\n 开始抽取标准表 \r\n "
+ DateForm.date2StringBysecond(new Date()) + DateForm.date2StringBysecond(new Date())
+ "\n\r\n"); + "\n\r\n");
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String string : rList) for (String string : rList)
sb.append(string).append("\n"); sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString()); Configs.CONSOLE_LOGGER.info(sb.toString());
// client.updateOrAddReplicasLabelById(collectOracle.getName(), // client.updateOrAddReplicasLabelById(collectOracle.getName(),
// "isExtract", "1"); // "isExtract", "1");
// //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
oracleExtract.createStandardDBLink(conn, collectOracle); // 创建dblink oracleExtract.createStandardDBLink(conn, collectOracle); // 创建dblink
oracleExtract.createStardardTableSpace(conn, collectOracle, oracleExtract.createStardardTableSpace(conn, collectOracle,
oracleModel); // 创建表空间 oracleModel); // 创建表空间
oracleExtract.createOnlyUser(conn, collectOracle, oracleExtract.createOnlyUser(conn, collectOracle,
oracleModel);// 创建 抽取标准表的 用户并授权 oracleModel);// 创建 抽取标准表的 用户并授权
DataInfoEntity tmpdata = dataInfoDao.findById(data.getId()); DataInfoEntity tmpdata = dataInfoDao.findById(data.getId());
if (null != tmpdata) { if (null != tmpdata) {
if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata
.getPayResultLast()) .getPayResultLast())
|| Constant.CHECKOUT_STATUS_FIVE.equals(tmpdata || Constant.CHECKOUT_STATUS_FIVE.equals(tmpdata
.getPayResultLast())) { .getPayResultLast())) {
// 抽取中 // 抽取中
data.setPayResultLast(Constant.CHECKOUT_STATUS_SIX); data.setPayResultLast(Constant.CHECKOUT_STATUS_SIX);
dataInfoDao.update(data); dataInfoDao.update(data);
boolean isExtrac = true; boolean isExtrac = true;
try { try {
oracleExtract.extractStandardPayTable(conn, oracleExtract.extractStandardPayTable(conn,
collectOracle, oracleModel);// 执行抽取 collectOracle, oracleModel);// 执行抽取
} catch (Exception e) { } catch (Exception e) {
// 改回 校验存在的状态 // 改回 校验存在的状态
data.setPayResultLast(Constant.CHECKOUT_STATUS_THREE); data.setPayResultLast(Constant.CHECKOUT_STATUS_THREE);
dataInfoDao.update(data); dataInfoDao.update(data);
isExtrac = false; isExtrac = false;
} }
if (isExtrac) { if (isExtrac) {
// 抽取成功 // 抽取成功
data.setPayResultLast(Constant.CHECKOUT_STATUS_SEVEN); data.setPayResultLast(Constant.CHECKOUT_STATUS_SEVEN);
dataInfoDao.update(data); dataInfoDao.update(data);
} }
} }
if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata
.getExecResultLast()) .getExecResultLast())
|| Constant.CHECKOUT_STATUS_FIVE.equals(tmpdata || Constant.CHECKOUT_STATUS_FIVE.equals(tmpdata
.getExecResultLast())) { .getExecResultLast())) {
// 抽取中 // 抽取中
data.setExecResultLast(Constant.CHECKOUT_STATUS_SIX); data.setExecResultLast(Constant.CHECKOUT_STATUS_SIX);
dataInfoDao.update(data); dataInfoDao.update(data);
boolean isExtrac = true; boolean isExtrac = true;
try { try {
oracleExtract.extractStandardExecTable(conn, oracleExtract.extractStandardExecTable(conn,
collectOracle, oracleModel);// 执行抽取 collectOracle, oracleModel);// 执行抽取
} catch (Exception e) { } catch (Exception e) {
// 改回 校验存在的状态 // 改回 校验存在的状态
data.setExecResultLast(Constant.CHECKOUT_STATUS_THREE); data.setExecResultLast(Constant.CHECKOUT_STATUS_THREE);
dataInfoDao.update(data); dataInfoDao.update(data);
isExtrac = false; isExtrac = false;
} }
if (isExtrac) { if (isExtrac) {
data.setExecResultLast(Constant.CHECKOUT_STATUS_SEVEN); data.setExecResultLast(Constant.CHECKOUT_STATUS_SEVEN);
dataInfoDao.update(data); dataInfoDao.update(data);
} }
} }
// client.updateOrAddReplicasLabelById(collectOracle.getName(), // client.updateOrAddReplicasLabelById(collectOracle.getName(),
// "isExtract", "2"); // "isExtract", "2");
// //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + replicasName + " annotate --overwrite rc " + replicasName
+ " standardExtractStatus=2"; + " standardExtractStatus=2";
rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
sb = new StringBuffer(); sb = new StringBuffer();
for (String string : rList) for (String string : rList)
sb.append(string).append("\n"); sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString()); Configs.CONSOLE_LOGGER.info(sb.toString());
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SEVEN); data.setCheckoutFlag(Constant.CHECKOUTFLAG_SEVEN);
data.setStandardExtractStatus("2"); data.setStandardExtractStatus("2");
dataInfoDao.update(data); dataInfoDao.update(data);
DataInfoEntity tmpSrcData = dataInfoDao.findById(data DataInfoEntity tmpSrcData = dataInfoDao.findById(data
.getId()); .getId());
data.setId(tmpSrcData.getSrcId()); data.setId(tmpSrcData.getSrcId());
dataInfoDao.update(data); dataInfoDao.update(data);
} }
} }
} catch (Exception e) { } catch (Exception e) {
String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + replicasName + " annotate --overwrite rc " + replicasName
+ " standardExtractStatus=0"; + " standardExtractStatus=0";
Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
data.setStandardExtractStatus("0"); data.setStandardExtractStatus("0");
dataInfoDao.update(data); dataInfoDao.update(data);
DataInfoEntity tmpSrcData = dataInfoDao.findById(data.getId()); DataInfoEntity tmpSrcData = dataInfoDao.findById(data.getId());
data.setId(tmpSrcData.getSrcId()); data.setId(tmpSrcData.getSrcId());
dataInfoDao.update(data); dataInfoDao.update(data);
log.error(Custom4exception.OracleSQL_Except, e); log.error(Custom4exception.OracleSQL_Except, e);
} finally { } finally {
// 去掉保存的当前数据id, // 去掉保存的当前数据id,
CacheSetCantDelete.removeStandardId(collectOracle.getDataId()); CacheSetCantDelete.removeStandardId(collectOracle.getDataId());
String msg = "抽取标准表结束"; String msg = "抽取标准表结束";
if (!"2".equals(data.getStandardExtractStatus())) { if (!"2".equals(data.getStandardExtractStatus())) {
msg += " 抽取有异常！状态重置为待抽取 "; msg += " 抽取有异常！状态重置为待抽取 ";
data.setStandardExtractStatus("0"); data.setStandardExtractStatus("0");
dataInfoDao.update(data); dataInfoDao.update(data);
DataInfoEntity tmpdata = dataInfoDao.findById(data.getId()); DataInfoEntity tmpdata = dataInfoDao.findById(data.getId());
data.setId(tmpdata.getSrcId()); data.setId(tmpdata.getSrcId());
dataInfoDao.update(data); dataInfoDao.update(data);
} }
// sql日志记录时间： // sql日志记录时间：
// sql日志记录时间： // sql日志记录时间：
FileOperateHelper.fileWrite( FileOperateHelper.fileWrite(
Configs.EXTRACT_STANDARD_LOG_LOCALTION Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ collectOracle.getName() + ".log", + collectOracle.getName() + ".log",
" " + msg + " >>>>> " " " + msg + " >>>>> "
+ DateForm.date2StringBysecond(new Date()) + DateForm.date2StringBysecond(new Date())
+ "\r\n\r\n\n"); + "\r\n\r\n\n");
} }
} }
isSuccess = true; isSuccess = true;
return isSuccess; return isSuccess;
} }
@Override @Override
public void updateDataExtractStatus(OracleConnectorParams ocp, int status) { public void updateDataExtractStatus(OracleConnectorParams ocp, int status) {
DataInfoEntity data = new DataInfoEntity(); DataInfoEntity data = new DataInfoEntity();
data.setId(Integer.valueOf(ocp.getDataId())); data.setId(Integer.valueOf(ocp.getDataId()));
// 设置为 标准表 抽取中 // 设置为 标准表 抽取中
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX); data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX);
data.setExtractStatus(status); data.setExtractStatus(status);
try { try {
dataInfoDao.update(data); dataInfoDao.update(data);
} catch (Exception e) { } catch (Exception e) {
// TODO Auto-generated catch block // TODO Auto-generated catch block
e.printStackTrace(); e.printStackTrace();
} }
} }
// @Override // @Override
// public boolean extractOracle(String name, List<OracleConnectorParams> // public boolean extractOracle(String name, List<OracleConnectorParams>
// dataInfos, GatherOracleInfo oracleConnect) throws Exception { // dataInfos, GatherOracleInfo oracleConnect) throws Exception {
// boolean isSuccess = false; // boolean isSuccess = false;
// try{ // try{
// //map转 bean(汇总库信息-带tableName的) // //map转 bean(汇总库信息-带tableName的)
// // GatherOracleInfo oracleModel = (GatherOracleInfo) // // GatherOracleInfo oracleModel = (GatherOracleInfo)
// Bean2MapUtils.convertMap(GatherOracleInfo.class, oracleConnect); // Bean2MapUtils.convertMap(GatherOracleInfo.class, oracleConnect);
// //
// //采集库连接参数 // //采集库连接参数
// // List<OracleConnectorParams> datainfos = new // // List<OracleConnectorParams> datainfos = new
// ArrayList<OracleConnectorParams>(); // ArrayList<OracleConnectorParams>();
// // for (Map<String, String> map : dataInfoMap) { // // for (Map<String, String> map : dataInfoMap) {
// // OracleConnectorParams dataInfoEntity = (OracleConnectorParams) // // OracleConnectorParams dataInfoEntity = (OracleConnectorParams)
// Bean2MapUtils.convertMap(OracleConnectorParams.class, oracleConnect); // Bean2MapUtils.convertMap(OracleConnectorParams.class, oracleConnect);
// // datainfos.add(dataInfoEntity); // // datainfos.add(dataInfoEntity);
// // } // // }
// //
// Connection conn = OracleConnector.ConnectionBuilder("jdbc:oracle:thin:@" // Connection conn = OracleConnector.ConnectionBuilder("jdbc:oracle:thin:@"
// + oracleConnect.getIp() + ":" + oracleConnect.getPort() + "/" // + oracleConnect.getIp() + ":" + oracleConnect.getPort() + "/"
// + oracleConnect.getDatabaseName(), oracleConnect.getUser(), // + oracleConnect.getDatabaseName(), oracleConnect.getUser(),
// oracleConnect.getPassword()); // oracleConnect.getPassword());
// //
// for (OracleConnectorParams collectOracle : dataInfos) { // for (OracleConnectorParams collectOracle : dataInfos) {
// //
// oracleExtract.createDBLink(conn, collectOracle); // oracleExtract.createDBLink(conn, collectOracle);
// oracleExtract.createTableSpace(conn, oracleConnect); // oracleExtract.createTableSpace(conn, oracleConnect);
// oracleExtract.createUser(conn, oracleConnect); // oracleExtract.createUser(conn, oracleConnect);
// oracleExtract.extractColleDB(conn, collectOracle); // oracleExtract.extractColleDB(conn, collectOracle);
// } // }
// isSuccess = true; // isSuccess = true;
// }catch(Exception e){ // }catch(Exception e){
// //
// } // }
// return false; // return false;
// } // }
} }

670
src/com/platform/service/thread/ThreadCheckoutStandardOracle.java
Unescape View File

@ -1,335 +1,335 @@
package com.platform.service.thread; package com.platform.service.thread;
import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.ReplicationController; import io.fabric8.kubernetes.api.model.ReplicationController;
import java.sql.Connection; import java.sql.Connection;
import java.util.List; import java.util.List;
import java.util.Set; import java.util.Set;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import com.platform.dao.DataInfoDao; import com.platform.dao.DataInfoDao;
import com.platform.entities.CheckoutEntity; import com.platform.entities.CheckoutEntity;
import com.platform.entities.DataInfoEntity; import com.platform.entities.DataInfoEntity;
import com.platform.entities.OracleConnectorParams; import com.platform.entities.OracleConnectorParams;
import com.platform.kubernetes.SimpleKubeClient; import com.platform.kubernetes.SimpleKubeClient;
import com.platform.oracle.OracleConnector; import com.platform.oracle.OracleConnector;
import com.platform.utils.CacheOracleCheckoutEntity; import com.platform.utils.CacheOracleCheckoutEntity;
import com.platform.utils.Configs; import com.platform.utils.Configs;
import com.platform.utils.Constant; import com.platform.utils.Constant;
import com.platform.utils.FileOperateHelper; import com.platform.utils.FileOperateHelper;
/** /**
* 校验oracle标准表是否存在 * 校验oracle标准表是否存在
* *
* @author chen * @author chen
* *
*/ */
public class ThreadCheckoutStandardOracle extends Thread { public class ThreadCheckoutStandardOracle extends Thread {
public final static Logger log = Configs.CONSOLE_LOGGER public final static Logger log = Configs.CONSOLE_LOGGER
.getLogger(ThreadCheckoutStandardOracle.class); .getLogger(ThreadCheckoutStandardOracle.class);
/** /**
* kuber 客户端 * kuber 客户端
*/ */
private SimpleKubeClient client; private SimpleKubeClient client;
/** /**
* dataInfo 的数据持久层 * dataInfo 的数据持久层
*/ */
private DataInfoDao dataInfoDao; private DataInfoDao dataInfoDao;
public ThreadCheckoutStandardOracle(DataInfoDao dataInfoDao) { public ThreadCheckoutStandardOracle(DataInfoDao dataInfoDao) {
this.setDaemon(true); this.setDaemon(true);
this.client = new SimpleKubeClient(); this.client = new SimpleKubeClient();
this.dataInfoDao = dataInfoDao; this.dataInfoDao = dataInfoDao;
} }
@Override @Override
public void run() { public void run() {
try { try {
Thread.sleep(1000 * 5); Thread.sleep(1000 * 5);
} catch (InterruptedException e2) { } catch (InterruptedException e2) {
log.error(e2); log.error(e2);
} }
// 循环11次，每次休眠 // 循环11次，每次休眠
for (int i = 0; i < 11; i++) { for (int i = 0; i < 11; i++) {
try { try {
// 数据的 keys （=kuber的应用名称 taskName) // 数据的 keys （=kuber的应用名称 taskName)
Set<String> taskNames = CacheOracleCheckoutEntity Set<String> taskNames = CacheOracleCheckoutEntity
.getCheckKeys(); .getCheckKeys();
int lengs = taskNames.size(); int lengs = taskNames.size();
if (lengs == 0) { if (lengs == 0) {
break; break;
} }
String[] taskNamekeys = taskNames.toArray(new String[lengs]); String[] taskNamekeys = taskNames.toArray(new String[lengs]);
StringBuffer sbtask = new StringBuffer(); StringBuffer sbtask = new StringBuffer();
for (int j = 0; j < taskNamekeys.length; j++) { for (int j = 0; j < taskNamekeys.length; j++) {
sbtask.append(taskNamekeys[j]).append("\t"); sbtask.append(taskNamekeys[j]).append("\t");
} }
log.info("replicationController标签： " + sbtask.toString()); log.info("replicationController标签： " + sbtask.toString());
for (String key : taskNamekeys) { for (String key : taskNamekeys) {
// 获得 kuber的 pod // 获得 kuber的 pod
Pod tmpPod = filterPod(key); Pod tmpPod = filterPod(key);
if (null == tmpPod) { if (null == tmpPod) {
log.info("replicationController标签： " + key log.info("replicationController标签： " + key
+ " 的 pod 节点不存在！"); + " 的 pod 节点不存在！");
if (i > 5) { if (i > 5) {
CacheOracleCheckoutEntity.checkRemove(key); CacheOracleCheckoutEntity.checkRemove(key);
} }
continue; continue;
} }
// 尝试 连接 oracle // 尝试 连接 oracle
connectOracle(tmpPod, key); connectOracle(tmpPod, key);
if (i == 10) { if (i == 10) {
String cmd = "kubectl --server " String cmd = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + key + " status=1"; + " label --overwrite rc " + key + " status=1";
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String str : rList) for (String str : rList)
sb.append(str).append("\n"); sb.append(str).append("\n");
log.info("更新replicationController标签： " + key log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 失败]"); + "\t[标签更新为： 失败]");
log.info(sb.toString()); log.info(sb.toString());
String cmd2 = "kubectl --server " String cmd2 = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + key + " annotate --overwrite rc " + key
+ " checkoutFlag=0"; + " checkoutFlag=0";
List<String> rList2 = Constant.ganymedSSH List<String> rList2 = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd2); .execCmdWaitAcquiescent(cmd2);
StringBuffer sb2 = new StringBuffer(); StringBuffer sb2 = new StringBuffer();
for (String str : rList2) for (String str : rList2)
sb2.append(str).append("\n"); sb2.append(str).append("\n");
log.info("更新replicationController标签： " + key log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 未校验]"); + "\t[标签更新为： 未校验]");
log.info(sb2.toString()); log.info(sb2.toString());
// 当前key标签对应的 数据服务的记录 // 当前key标签对应的 数据服务的记录
CheckoutEntity tmp = CacheOracleCheckoutEntity CheckoutEntity tmp = CacheOracleCheckoutEntity
.getCheck(key); .getCheck(key);
tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ZERO); tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ZERO);
tmp.setPayResultLast(Constant.CHECKOUT_STATUS_ZERO); tmp.setPayResultLast(Constant.CHECKOUT_STATUS_ZERO);
tmp.setExecResultLast(Constant.CHECKOUT_STATUS_ZERO); tmp.setExecResultLast(Constant.CHECKOUT_STATUS_ZERO);
try { try {
// 更新数据库 // 更新数据库
this.updateDataInfo(tmp); this.updateDataInfo(tmp);
} catch (Exception e) { } catch (Exception e) {
log.error(e); log.error(e);
} }
} }
} }
} catch (Exception e1) { } catch (Exception e1) {
log.error(e1.getStackTrace()); log.error(e1.getStackTrace());
} finally { } finally {
try { try {
Thread.sleep(1000 * 60); Thread.sleep(1000 * 60);
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.error(e); log.error(e);
} }
} }
} }
} }
/** /**
* 查找 pod 节点 * 查找 pod 节点
* *
* @param taskName * @param taskName
* @return * @return
*/ */
private Pod filterPod(String taskName) { private Pod filterPod(String taskName) {
Pod pod = null; Pod pod = null;
ReplicationController replicationController = client ReplicationController replicationController = client
.getReplicationController(taskName); .getReplicationController(taskName);
if (null != replicationController) { if (null != replicationController) {
List<Pod> filterPods = client List<Pod> filterPods = client
.getPodsForApplicaList(replicationController); .getPodsForApplicaList(replicationController);
if (filterPods != null && filterPods.size() > 0) { if (filterPods != null && filterPods.size() > 0) {
pod = filterPods.get(0); pod = filterPods.get(0);
} }
} }
return pod; return pod;
} }
/** /**
* 尝试连接 oracle服务 * 尝试连接 oracle服务
* *
* @param tmpPod * @param tmpPod
* @param key * @param key
*/ */
private void connectOracle(Pod pod, String key) throws Exception { private void connectOracle(Pod pod, String key) throws Exception {
if (pod != null) { if (pod != null) {
String ip = client.getPodHostIp(pod); String ip = client.getPodHostIp(pod);
int port = client.getPodContainerport(pod); int port = client.getPodContainerport(pod);
if (ip != null && port != 0) { if (ip != null && port != 0) {
String url = "jdbc:oracle:thin:@" + ip + ":" + port + ":" String url = "jdbc:oracle:thin:@" + ip + ":" + port + ":"
+ Configs.ORACLE_ORCL; + Configs.ORACLE_ORCL;
boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败 boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败
Configs.ORACLE_USER, Configs.ORACLE_PSW); Configs.ORACLE_USER, Configs.ORACLE_PSW);
log.info("url:" + url + ",user:" + Configs.ORACLE_USER log.info("url:" + url + ",user:" + Configs.ORACLE_USER
+ ",password:" + Configs.ORACLE_PSW); + ",password:" + Configs.ORACLE_PSW);
String message = "失败"; String message = "失败";
String cmd3 = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS String cmd3 = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + key + " checkoutFlag=0"; + " annotate --overwrite rc " + key + " checkoutFlag=0";
List<String> rList3 = Constant.ganymedSSH List<String> rList3 = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd3); .execCmdWaitAcquiescent(cmd3);
StringBuffer sb3 = new StringBuffer(); StringBuffer sb3 = new StringBuffer();
for (String str : rList3) for (String str : rList3)
sb3.append(str).append("\n"); sb3.append(str).append("\n");
log.info(sb3.toString()); log.info(sb3.toString());
log.info("更新replicationController标签： " + key + "\t[标签更新为：未校验]"); log.info("更新replicationController标签： " + key + "\t[标签更新为：未校验]");
if (flag) { if (flag) {
String cmd = "kubectl --server " String cmd = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + key + " status=2"; + " label --overwrite rc " + key + " status=2";
// 设置服务为 成功 // 设置服务为 成功
List<String> rList = Constant.ganymedSSH List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd); .execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String str : rList) for (String str : rList)
sb.append(str).append("\n"); sb.append(str).append("\n");
log.info(sb.toString()); log.info(sb.toString());
message = "成功"; message = "成功";
log.info("更新replicationController标签： " + key log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 成功]"); + "\t[标签更新为： 成功]");
// 校验标签 // 校验标签
String cmd2 = "kubectl --server " String cmd2 = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + key + " annotate --overwrite rc " + key
+ " checkoutFlag=2"; + " checkoutFlag=2";
List<String> rList2 = Constant.ganymedSSH List<String> rList2 = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd2); .execCmdWaitAcquiescent(cmd2);
StringBuffer sb2 = new StringBuffer(); StringBuffer sb2 = new StringBuffer();
for (String str : rList2) for (String str : rList2)
sb2.append(str).append("\n"); sb2.append(str).append("\n");
log.info(sb2.toString()); log.info(sb2.toString());
log.info("更新replicationController标签： " + key log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 校验中]"); + "\t[标签更新为： 校验中]");
// 获得当前 服务对应的 数据 // 获得当前 服务对应的 数据
CheckoutEntity tmp = CacheOracleCheckoutEntity CheckoutEntity tmp = CacheOracleCheckoutEntity
.getCheck(key); .getCheck(key);
if (null != tmp) { if (null != tmp) {
CacheOracleCheckoutEntity.putExtract(key, tmp); CacheOracleCheckoutEntity.putExtract(key, tmp);
} }
// 查询 对应的 2 个标准表 // 查询 对应的 2 个标准表
OracleConnectorParams oc = new OracleConnectorParams(); OracleConnectorParams oc = new OracleConnectorParams();
String logName = tmp.getAreaCode().toLowerCase() + "_" String logName = tmp.getAreaCode().toLowerCase() + "_"
+ tmp.getSysCode() + "_" + tmp.getDataVersion(); + tmp.getSysCode() + "_" + tmp.getDataVersion();
oc.setName(logName); oc.setName(logName);
try { try {
Connection conn = OracleConnector.connectionBuilder( Connection conn = OracleConnector.connectionBuilder(
url, Configs.ORACLE_USER, Configs.ORACLE_PSW, url, Configs.ORACLE_USER, Configs.ORACLE_PSW,
oc); oc);
// 支付--校验 // 支付--校验
if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp
.getPayResultLast())) { .getPayResultLast())) {
String paySql = "select * from dba_tables where owner = '" String paySql = "select * from dba_tables where owner = '"
+ Configs.COLLECT_STANDARD_TABLE_USER + Configs.COLLECT_STANDARD_TABLE_USER
.toUpperCase() .toUpperCase()
+ "' and table_name = '" + "' and table_name = '"
+ Configs.COLLECT_PAY_TABLE.toUpperCase() + Configs.COLLECT_PAY_TABLE.toUpperCase()
+ "'"; + "'";
if (OracleConnector if (OracleConnector
.execUpdateOracleSQL( .execUpdateOracleSQL(
conn, conn,
paySql, paySql,
FileOperateHelper FileOperateHelper
.addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION) .addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION)
+ logName + "jy.log")) { + logName + "jy.log")) {
tmp.setPayResultLast(Constant.CHECKOUT_STATUS_THREE); tmp.setPayResultLast(Constant.CHECKOUT_STATUS_THREE);
} else { } else {
tmp.setPayResultLast(Constant.CHECKOUT_STATUS_FOUR); tmp.setPayResultLast(Constant.CHECKOUT_STATUS_FOUR);
} }
} }
// 可执行-- 校验 // 可执行-- 校验
if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp
.getExecResultLast())) { .getExecResultLast())) {
String execSql = "select * from dba_tables where owner = '" String execSql = "select * from dba_tables where owner = '"
+ Configs.COLLECT_STANDARD_TABLE_USER + Configs.COLLECT_STANDARD_TABLE_USER
.toUpperCase() .toUpperCase()
+ "' and table_name = '" + "' and table_name = '"
+ Configs.COLLECT_EXEC_TABLE.toUpperCase() + Configs.COLLECT_EXEC_TABLE.toUpperCase()
+ "'"; + "'";
if (OracleConnector if (OracleConnector
.execUpdateOracleSQL( .execUpdateOracleSQL(
conn, conn,
execSql, execSql,
FileOperateHelper FileOperateHelper
.addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION) .addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION)
+ logName + "jy.log")) { + logName + "jy.log")) {
tmp.setExecResultLast(Constant.CHECKOUT_STATUS_THREE); tmp.setExecResultLast(Constant.CHECKOUT_STATUS_THREE);
} else { } else {
tmp.setExecResultLast(Constant.CHECKOUT_STATUS_FOUR); tmp.setExecResultLast(Constant.CHECKOUT_STATUS_FOUR);
} }
} }
tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ONE); tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ONE);
// 更新数据库data_info // 更新数据库data_info
updateDataInfo(tmp); updateDataInfo(tmp);
// 将 CacheOracleCheckoutEntity checkMap中的 数据 // 将 CacheOracleCheckoutEntity checkMap中的 数据
// 放入extractMap中 // 放入extractMap中
/* CacheOracleCheckoutEntity.putExtract(key, */CacheOracleCheckoutEntity /* CacheOracleCheckoutEntity.putExtract(key, */CacheOracleCheckoutEntity
.checkRemove(key)/* ) */; .checkRemove(key)/* ) */;
// 更新kuber状态 // 更新kuber状态
cmd2 = "kubectl --server " cmd2 = "kubectl --server "
+ Configs.KUBE_MASTER_ADDRESS + Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + key + " annotate --overwrite rc " + key
+ " checkoutFlag=1"; + " checkoutFlag=1";
// client.updateOrAddReplicasLabelById(taskNSyame, // client.updateOrAddReplicasLabelById(taskNSyame,
// "status", "2"); // "status", "2");
rList = Constant.ganymedSSH rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd2); .execCmdWaitAcquiescent(cmd2);
sb = new StringBuffer(); sb = new StringBuffer();
for (String str : rList) for (String str : rList)
sb.append(str).append("\n"); sb.append(str).append("\n");
log.info(sb.toString()); log.info(sb.toString());
message = "成功"; message = "成功";
log.info("更新replicationController标签： " + key log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 已校验]"); + "\t[标签更新为： 已校验]");
} catch (Exception e) { } catch (Exception e) {
log.error(e); log.error(e);
} }
// 成功 就 清除 CacheOracleCheckoutEntity 中 的该条记录 // 成功 就 清除 CacheOracleCheckoutEntity 中 的该条记录
CacheOracleCheckoutEntity.checkRemove(key); CacheOracleCheckoutEntity.checkRemove(key);
} }
log.info("连接到数据库服务： " + key + "\t[连接结果： " + message + "]"); log.info("连接到数据库服务： " + key + "\t[连接结果： " + message + "]");
} }
} }
} }
/** /**
* 更新 dataInfo表 * 更新 dataInfo表
* *
* @param checkoutEntity * @param checkoutEntity
* @throws Exception * @throws Exception
*/ */
private void updateDataInfo(CheckoutEntity checkoutEntity) throws Exception { private void updateDataInfo(CheckoutEntity checkoutEntity) throws Exception {
DataInfoEntity data = new DataInfoEntity(); DataInfoEntity data = new DataInfoEntity();
// 状态改为 正则校验 Constant.CHECKOUTFLAG_TWO // 状态改为 正则校验 Constant.CHECKOUTFLAG_TWO
data.setId(checkoutEntity.getDataId()); data.setId(checkoutEntity.getDataId());
data.setPayResultLast(checkoutEntity.getPayResultLast()); data.setPayResultLast(checkoutEntity.getPayResultLast());
data.setExecResultLast(checkoutEntity.getExecResultLast()); data.setExecResultLast(checkoutEntity.getExecResultLast());
data.setCheckoutFlag(checkoutEntity.getCheckoutFlag()); data.setCheckoutFlag(checkoutEntity.getCheckoutFlag());
dataInfoDao.update(data); dataInfoDao.update(data);
DataInfoEntity tmpdata = dataInfoDao.findById(checkoutEntity DataInfoEntity tmpdata = dataInfoDao.findById(checkoutEntity
.getDataId()); .getDataId());
data.setId(tmpdata.getSrcId()); data.setId(tmpdata.getSrcId());
dataInfoDao.update(data); dataInfoDao.update(data);
} }
} }

Write Preview
Loading…
Cancel
Save