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

----合并

Browse Source
web_backend_develope
wu ming 8 years ago
parent 83b3c8975a 1fd5b15305
commit 7bcc95e0d4
16 changed files with 1811 additions and 1695 deletions
Show Stats Download Patch File Download Diff File

239
WebContent/WEB-INF/config/config.properties
Unescape View File

@ -1,119 +1,120 @@
#============================================================================================================= #=============================================================================================================
# MySQL # MySQL
#============================================================================================================= #=============================================================================================================
# 驱动程序 # 驱动程序
jdbc.mysql.driver=com.mysql.jdbc.Driver jdbc.mysql.driver=com.mysql.jdbc.Driver
# 连接url # 连接url
jdbc.mysql.url=jdbc:mysql://192.168.0.110:3306/ftpdata?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&autoReconnect=true&failOverReadOnly=false jdbc.mysql.url=jdbc:mysql://192.168.0.110:3306/ftpdata?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&autoReconnect=true&failOverReadOnly=false
# 用户名 # 用户名
jdbc.mysql.username=root jdbc.mysql.username=root
# 密码 # 密码
jdbc.mysql.password=root jdbc.mysql.password=root
#============================================================================================================= #=============================================================================================================
# Oracle 连接配置 # Oracle 连接配置
#============================================================================================================= #=============================================================================================================
# hui用户名 # hui用户名
gather-user-name=system gather-user-name=system
gather-standard-user-name=u_bzbjy gather-standard-user-name=u_bzbjy
gather-standard-exec-table-name=kzxzb gather-standard-exec-table-name=kzxzb
gather-standard-pay-table-name=zfxxb gather-standard-pay-table-name=zfxxb
# 密码 # 密码
gather-user-password=oracle gather-user-password=oracle
# #
gather-port=1521 gather-port=1521
# #
gather-service-name=orcl gather-service-name=orcl
#============================================================================================================= #=============================================================================================================
# 数据库公共配置 # 数据库公共配置
#============================================================================================================= #=============================================================================================================
jdbc.initialSize=5 jdbc.initialSize=5
jdbc.minIdle=5 jdbc.minIdle=5
jdbc.maxIdle=20 jdbc.maxIdle=20
jdbc.maxActive=100 jdbc.maxActive=100
jdbc.maxWait=100000 jdbc.maxWait=100000
jdbc.defaultAutoCommit=false jdbc.defaultAutoCommit=false
jdbc.removeAbandoned=true jdbc.removeAbandoned=true
jdbc.removeAbandonedTimeout=600 jdbc.removeAbandonedTimeout=600
jdbc.testWhileIdle=true jdbc.testWhileIdle=true
jdbc.timeBetweenEvictionRunsMillis=60000 jdbc.timeBetweenEvictionRunsMillis=60000
jdbc.numTestsPerEvictionRun=20 jdbc.numTestsPerEvictionRun=20
jdbc.minEvictableIdleTimeMillis=300000 jdbc.minEvictableIdleTimeMillis=300000
#============================================================================================================== #==============================================================================================================
# #
#============================================================================================================== #==============================================================================================================
table-suffix=_20152016 table-suffix=_20152016
gather-tablespace-name=TS_TTSSS gather-tablespace-name=TS_TTSSS
#dbf file's path must exist #dbf file's path must exist
gather-tablespace-path=/opt/oracle/app/oradata/orcl/ gather-tablespace-path=/opt/oracle/app/oradata/orcl/
# user's password must exist # user's password must exist
gather-table-user-password=user gather-table-user-password=user
#kuber's url must exist #kuber's url must exist
kubeMasterUrl=http://192.168.0.110:8080/ kubeMasterUrl=http://192.168.0.110:8080/
kubeMasterAddress=192.168.0.110:8080
kubeMasterAddress=192.168.0.110:8080
collect-user-name=system
collect-password=oracle collect-user-name=system
collect-service-name=orcl collect-password=oracle
# standard table : collect-service-name=orcl
# eg:sql -> select count(*) from u_bzbjy.kzxzb; select count(*) from u_bzbjy.zfxxb; # standard table :
collect-user-table=U_BZBJY # eg:sql -> select count(*) from u_bzbjy.kzxzb; select count(*) from u_bzbjy.zfxxb;
collect-exec-table=KZXZB collect-user-table=U_BZBJY
collect-pay-table=ZFXXB collect-exec-table=KZXZB
#gfs web's IP collect-pay-table=ZFXXB
gfs_control_ip=192.168.0.110 #gfs web's IP
#gfs web password of root gfs_control_ip=192.168.0.110
gfs_control_rootPassWd=root #gfs web password of root
#url of gfs web gfs_control_rootPassWd=root
HttpClientConstant_URL_IP_PORT=http://192.168.0.110:9001/ #url of gfs web
#data between 6 month ago and now HttpClientConstant_URL_IP_PORT=http://192.168.0.110:9001/
dataBefore=6 #data between 6 month ago and now
#move a data once dataBefore=6
moveFileMaxNum=1 #move a data once
#import 5 rows of Excel once moveFileMaxNum=1
numOneImportExcel=5 #import 5 rows of Excel once
oracle-orcl=orcl numOneImportExcel=5
oracle-user=system oracle-orcl=orcl
oracle-psw=oracle oracle-user=system
oracle-psw=oracle
#=============================================================================================================
# windows path #=============================================================================================================
#============================================================================================================= # windows path
#=============================================================================================================
extract-log-localtion=D:\\test\\log\\
extract-standard-log-localtion=D:\\test\\log2\\ extract-log-localtion=D:\\test\\log\\
extract-standard-log-localtion=D:\\test\\log2\\
file_upload_path=D:\\test\\
file_download_path=D:\\test\\export.xlsx file_upload_path=D:\\test\\
file_download_path=D:\\test\\export.xlsx
package_download_path=D:\\test\\
package_name=sql_script_standard package_download_path=D:\\test\\
package_name=sql_script_standard
sql_script_path_last=D:\\test\\sql_script_last\\
sql_script_path_standard=D:\\test\\sql_script_standard\\ sql_script_path_last=D:\\test\\sql_script_last\\
sql_script_path_standard=D:\\test\\sql_script_standard\\
#=============================================================================================================
# linux path #=============================================================================================================
#============================================================================================================= # linux path
## extract's log file location #=============================================================================================================
#extract-log-localtion=/home/web_manage/log/ ## extract's log file location
## standard extract's log file location #extract-log-localtion=/home/web_manage/log/
#extract-standard-log-localtion=/home/web_manage/log2/ ## standard extract's log file location
##import excel file's location #extract-standard-log-localtion=/home/web_manage/log2/
#file_upload_path=/excel_import_dir/ ##import excel file's location
##export excel file's location #file_upload_path=/excel_import_dir/
#file_download_path=/excel_export_dir/export.xlsx ##export excel file's location
##file package download's location #file_download_path=/excel_export_dir/export.xlsx
#package_download_path=/ ##file package download's location
##file package's name #package_download_path=/
#package_name=DefaultDescription ##file package's name
##location of sql script lastest #package_name=DefaultDescription
#sql_script_path_last=/DefaultDescription_last/ ##location of sql script lastest
##location of sql script standard #sql_script_path_last=/DefaultDescription_last/
#sql_script_path_standard=/DefaultDescription/ ##location of sql script standard
#sql_script_path_standard=/DefaultDescription/

6
src/com/platform/controller/CheckoutController.java
Unescape View File

@ -58,7 +58,7 @@ public class CheckoutController extends BaseController {
} }
/** /**
* 信息系统--校验--查看所有系统的 标准表情况 * 信息系统--校验--查看系统的 标准表情况
* @return * @return
* @throws Exception * @throws Exception
*/ */
@ -95,7 +95,7 @@ public class CheckoutController extends BaseController {
} }
/** /**
* 信息系统--校验--查看单条数据的修改详情 * 信息系统--校验--查看单条数据的详情
* @return * @return
* @throws Exception * @throws Exception
*/ */
@ -116,7 +116,7 @@ public class CheckoutController extends BaseController {
} }
/** /**
* 信息系统--校验--查看后 -> 单条数据的修改详情 * 信息系统--校验--(查看后 )单条数据的修改详情
* @return * @return
* @throws Exception * @throws Exception
*/ */

974
src/com/platform/controller/DataModelController.java
Unescape View File

@ -1,473 +1,501 @@
package com.platform.controller; package com.platform.controller;
import java.io.UnsupportedEncodingException; import java.io.UnsupportedEncodingException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import javax.annotation.Resource; import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse; import javax.servlet.http.HttpServletResponse;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import org.springframework.stereotype.Controller; import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap; import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseBody;
import com.base.BaseController; import com.base.BaseController;
import com.platform.entities.BasedTask; import com.platform.entities.BasedTask;
import com.platform.entities.DataInfoEntity; import com.platform.entities.DataInfoEntity;
import com.platform.entities.DataInfoEntityMoveTmp; import com.platform.entities.DataInfoEntityMoveTmp;
import com.platform.entities.GatherOracleInfo; import com.platform.entities.GatherOracleInfo;
import com.platform.entities.OracleConnectorParams; import com.platform.entities.OracleConnectorParams;
import com.platform.entities.OracleExtractExecuter; import com.platform.entities.OracleExtractExecuter;
import com.platform.form.PagerOptions; import com.platform.form.PagerOptions;
import com.platform.form.oracleForm; import com.platform.form.oracleForm;
import com.platform.form.volumeMoveForm; import com.platform.form.volumeMoveForm;
import com.platform.service.DataInfoService; import com.platform.service.DataInfoService;
import com.platform.service.ICodeService; import com.platform.service.ICodeService;
import com.platform.service.IGatherOracleService; import com.platform.service.IGatherOracleService;
import com.platform.service.ILogRead; import com.platform.service.ILogRead;
import com.platform.service.IMoveDataService; import com.platform.service.IMoveDataService;
import com.platform.service.IOracleExtractService; import com.platform.service.IOracleExtractService;
import com.platform.service.IVolumeService; import com.platform.service.IVolumeService;
import com.platform.service.OracleExtractTask; import com.platform.service.OracleExtractTask;
import com.platform.service.OracleStatusService; import com.platform.service.OracleStatusService;
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.ThreadVolumeImm; import com.platform.utils.ThreadVolumeImm;
import com.platform.utils.UtilsHelper; import com.platform.utils.UtilsHelper;
/** /**
* 数据管理 * 数据管理
* *
* @author chen * @author chen
* *
*/ */
@Controller @Controller
public class DataModelController extends BaseController { public class DataModelController extends BaseController {
public static Logger log = Logger.getLogger(DataModelController.class); public static Logger log = Logger.getLogger(DataModelController.class);
/** /**
* 数据管理--业务类 * 数据管理--业务类
*/ */
@Resource(name = "dataInfoService") @Resource(name = "dataInfoService")
private DataInfoService dfs; private DataInfoService dfs;
/** /**
* gfs业务类 * gfs业务类
*/ */
@Resource(name = "volumeService") @Resource(name = "volumeService")
private IVolumeService volumeService; private IVolumeService volumeService;
/** /**
* mysql相关业务 * mysql相关业务
*/ */
@Resource(name = "mySqlService") @Resource(name = "mySqlService")
private IGatherOracleService mySqlService; private IGatherOracleService mySqlService;
/** /**
* oracle汇总功能业务 * oracle汇总功能业务
*/ */
@Resource(name = "OracleExtract") @Resource(name = "OracleExtract")
private IOracleExtractService OracleExtract; private IOracleExtractService OracleExtract;
/** /**
* 迁移数据业务类 * 迁移数据业务类
*/ */
@Resource(name = "moveDataService") @Resource(name = "moveDataService")
private IMoveDataService moveDataService; private IMoveDataService moveDataService;
/** /**
* 地区、系统code业务类 * 地区、系统code业务类
*/ */
@Resource(name = "codeService") @Resource(name = "codeService")
private ICodeService codeService; private ICodeService codeService;
/** /**
* 日志业务管理类 * 日志业务管理类
*/ */
@Resource(name = "logReadService") @Resource(name = "logReadService")
private ILogRead logReadService; private ILogRead logReadService;
public void setDfsImp(DataInfoService dfs) { public void setDfsImp(DataInfoService dfs) {
this.dfs = dfs; this.dfs = dfs;
} }
/** /**
* 数据管理--分页查询数据 * 数据管理--分页查询数据
* *
* @param res * @param res
* @param req * @param req
* @return * @return
* @throws Exception * @throws Exception
*/ */
@RequestMapping("/data.json") @RequestMapping("/data.json")
@ResponseBody @ResponseBody
public ModelMap getAllDataToJson(HttpServletRequest res, public ModelMap getAllDataToJson(HttpServletRequest res,
HttpServletResponse req) throws Exception { HttpServletResponse req) throws Exception {
log.info("-----------/data.json-----------"); log.info("-----------/data.json-----------");
res.setCharacterEncoding("UTF-8"); res.setCharacterEncoding("UTF-8");
Map<String, String[]> paramMap = res.getParameterMap(); Map<String, String[]> paramMap = res.getParameterMap();
Set<String> keySet = paramMap.keySet(); Set<String> keySet = paramMap.keySet();
Map<String, String> params = new HashMap<String, String>(); Map<String, String> params = new HashMap<String, String>();
StringBuffer sb = new StringBuffer().append("当前的请求参数:{"); StringBuffer sb = new StringBuffer().append("当前的请求参数:{");
for (String str : keySet) { for (String str : keySet) {
String value = paramMap.get(str)[0]; String value = paramMap.get(str)[0];
if (StringUtils.isNotEmpty(value)) { if (StringUtils.isNotEmpty(value)) {
params.put(str, value); params.put(str, value);
sb.append(str).append(":").append(value).append(","); sb.append(str).append(":").append(value).append(",");
} else { } else {
sb.append(str).append(":").append("null").append(","); sb.append(str).append(":").append("null").append(",");
} }
} }
log.info(sb.deleteCharAt(sb.length() - 1).append("}").toString()); log.info(sb.deleteCharAt(sb.length() - 1).append("}").toString());
PagerOptions pagerOptions = (PagerOptions) UtilsHelper PagerOptions pagerOptions = (PagerOptions) UtilsHelper
.newObjAndSetAttrsByClass(PagerOptions.class, params); .newObjAndSetAttrsByClass(PagerOptions.class, params);
pagerOptions.setCurrentPageNum(Integer.valueOf(params pagerOptions.setCurrentPageNum(Integer.valueOf(params
.get("currentPageNum"))); .get("currentPageNum")));
// 冷热区查询字段mark // 冷热区查询字段mark
pagerOptions.setMark(pagerOptions.getVolumeType().trim()); pagerOptions.setMark(pagerOptions.getVolumeType().trim());
return dfs.getPagerTableData(pagerOptions); return dfs.getPagerTableData(pagerOptions);
} }
/** /**
* 数据管理--删除数据 * 数据管理--删除数据
* *
* @param res * @param res
* @param req * @param req
* @throws Exception * @throws Exception
*/ */
@RequestMapping(value = "/delete/data") @RequestMapping(value = "/delete/data")
@ResponseBody @ResponseBody
public ModelMap deleteData(HttpServletRequest res, HttpServletResponse req) public ModelMap deleteData(HttpServletRequest res, HttpServletResponse req)
throws Exception { throws Exception {
log.info("-----------/delete/data-----------"); log.info("-----------/delete/data-----------");
ModelMap mode = new ModelMap(); ModelMap mode = new ModelMap();
res.setCharacterEncoding("UTF-8"); res.setCharacterEncoding("UTF-8");
Map<String, String[]> paramMap = res.getParameterMap(); Map<String, String[]> paramMap = res.getParameterMap();
String[] data = paramMap.get("data"); String[] data = paramMap.get("data");
if (null != data && data.length > 0) { if (null != data && data.length > 0) {
List<String> list = new ArrayList<String>(); List<String> list = new ArrayList<String>();
List<String> errList = new ArrayList<String>(); List<String> errList = new ArrayList<String>();
// 判断是否有 不能删除的。 // 判断是否有 不能删除的。
for (String dataId : data) { for (String dataId : data) {
if (CacheSetCantDelete.containsId(dataId)) if (CacheSetCantDelete.containsId(dataId))
errList.add(dataId); errList.add(dataId);
else else
list.add(dataId); list.add(dataId);
} }
if (list.size() > 0) { if (list.size() > 0) {
String[] ids = list.toArray(new String[list.size()]); String[] ids = list.toArray(new String[list.size()]);
dfs.deleteData(ids); dfs.deleteData(ids);
req.setStatus(200); req.setStatus(200);
} }
if (errList.size() > 0) { if (errList.size() > 0) {
mode.put("err", errList); mode.put("err", errList);
req.setStatus(500); req.setStatus(500);
} }
} }
return mode; return mode;
} }
/** /**
* 连接oracle * 连接oracle
* *
* @param res * @param res
* @param req * @param req
* @throws UnsupportedEncodingException * @throws UnsupportedEncodingException
*/ */
@RequestMapping("/connectOracle") @RequestMapping("/connectOracle")
public void connectOracle(HttpServletRequest res, HttpServletResponse req) public void connectOracle(HttpServletRequest res, HttpServletResponse req)
throws UnsupportedEncodingException { throws UnsupportedEncodingException {
res.setCharacterEncoding("UTF-8"); res.setCharacterEncoding("UTF-8");
Map<String, String[]> paramMap = res.getParameterMap(); Map<String, String[]> paramMap = res.getParameterMap();
String[] oraclesName = paramMap.get("oracleName"); String[] oraclesName = paramMap.get("oracleName");
if (oraclesName != null) if (oraclesName != null)
for (String rcName : oraclesName) { for (String rcName : oraclesName) {
log.info("执行连接\t" + rcName); log.info("执行连接\t" + rcName);
String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
+ " label --overwrite rc " + rcName + " status=0"; String cmd = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
List<String> rList = Constant.ganymedSSH + " label --overwrite rc " + rcName + " status=0";
.execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); List<String> rList = Constant.ganymedSSH
for (String string : rList) .execCmdWaitAcquiescent(cmd);
sb.append(string).append("\n"); StringBuffer sb = new StringBuffer();
log.info(sb.toString()); for (String string : rList)
new OracleStatusService().connectToOracle(rcName); sb.append(string).append("\n");
} log.info(sb.toString());
req.setStatus(200); new OracleStatusService().connectToOracle(rcName);
} }
req.setStatus(200);
/** }
* 断开oracle连接
* /**
* @param res * 断开oracle连接
* @param req *
* @throws UnsupportedEncodingException * @param res
*/ * @param req
@RequestMapping("/cancelOracleConection") * @throws UnsupportedEncodingException
public void cancelOracleConnection(HttpServletRequest res, */
HttpServletResponse req) throws UnsupportedEncodingException { @RequestMapping("/cancelOracleConection")
res.setCharacterEncoding("UTF-8"); public void cancelOracleConnection(HttpServletRequest res,
Map<String, String[]> paramMap = res.getParameterMap(); HttpServletResponse req) throws UnsupportedEncodingException {
String[] oraclesName = paramMap.get("oracleName"); res.setCharacterEncoding("UTF-8");
String operate = paramMap.get("operation")[0]; Map<String, String[]> paramMap = res.getParameterMap();
if (null != oraclesName) { String[] oraclesName = paramMap.get("oracleName");
for (String rcName : oraclesName) { String operate = paramMap.get("operation")[0];
log.info("取消连接:\t" + rcName); if (null != oraclesName) {
new OracleStatusService().cancelToOracle(rcName, operate); for (String rcName : oraclesName) {
} log.info("取消连接:\t" + rcName);
} new OracleStatusService().cancelToOracle(rcName, operate);
req.setStatus(200); }
} }
req.setStatus(200);
/** }
* gfs的 volume节点的查询
* /**
* @return * gfs的 volume节点的查询
* @throws Exception *
*/ * @return
@RequestMapping(value = "/volume/list", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public String volumeList() throws Exception { @RequestMapping(value = "/volume/list", method = RequestMethod.POST)
log.info("-----/volume/list------"); @ResponseBody
String rest = volumeService.getAllvolume(); public String volumeList() throws Exception {
new ThreadVolumeImm("ThreadVolumeImm-in-VolumeController-volumeList") log.info("-----/volume/list------");
.start(); String rest = volumeService.getAllvolume();
return rest; new ThreadVolumeImm("ThreadVolumeImm-in-VolumeController-volumeList")
} .start();
return rest;
/** }
* 数据迁移功能
* /**
* @param res * 数据迁移功能
* @param req *
* @param form * @param res
* @throws Exception * @param req
*/ * @param form
@RequestMapping(value = "/volume/{name}/move", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public void volumeMove(HttpServletRequest res, HttpServletResponse req, @RequestMapping(value = "/volume/{name}/move", method = RequestMethod.POST)
@RequestBody volumeMoveForm form) throws Exception { @ResponseBody
List<DataInfoEntity> datas = new ArrayList<DataInfoEntity>(); public void volumeMove(HttpServletRequest res, HttpServletResponse req,
List<DataInfoEntity> selectItems = form.getSelectItems(); @RequestBody volumeMoveForm form) throws Exception {
if (null == selectItems) { List<DataInfoEntity> datas = new ArrayList<DataInfoEntity>();
req.setStatus(200); List<DataInfoEntity> selectItems = form.getSelectItems();
return; if (null == selectItems) {
} req.setStatus(200);
datas.addAll(selectItems); return;
}
log.info("------/volume/{name}/move--------"); datas.addAll(selectItems);
if (datas.size() > 0) {
moveDataService.moveData(datas, form.getSelectNode()); log.info("------/volume/{name}/move--------");
} if (datas.size() > 0) {
req.setStatus(200); moveDataService.moveData(datas, form.getSelectNode());
} }
req.setStatus(200);
/** }
* oracle的 汇总功能
* /**
* @param res * oracle的 汇总功能
* @param req *
* @param form * @param res
* @throws Exception * @param req
*/ * @param form
@RequestMapping(value = "/oracle/{name}/extract", method = RequestMethod.POST) * @throws Exception
public void oracleExtract(HttpServletRequest res, HttpServletResponse req, */
@RequestBody oracleForm form) throws Exception { @RequestMapping(value = "/oracle/{name}/extract", method = RequestMethod.POST)
log.info("/oracle/{name}/extract"); public void oracleExtract(HttpServletRequest res, HttpServletResponse req,
if (null != form.getTarget()) { // 检查请求参数中是否包含汇总库信息 @RequestBody oracleForm form) throws Exception {
boolean isConnect = OracleExtract.isConnectTotalOracle(form log.info("/oracle/{name}/extract");
.getTarget()); // 检查汇总库是否可以连接成功，连接成功返回200状态吗，连接失败返回500状态吗 if (null != form.getTarget()) { // 检查请求参数中是否包含汇总库信息
if (isConnect) { boolean isConnect = OracleExtract.isConnectTotalOracle(form
req.setStatus(200); .getTarget()); // 检查汇总库是否可以连接成功，连接成功返回200状态吗，连接失败返回500状态吗
if (null != form.getInneed() && form.getInneed().size() > 0) { if (isConnect) {
for (OracleConnectorParams oracleParams : form.getInneed()) { req.setStatus(200);
BasedTask task = new OracleExtractTask( if (null != form.getInneed() && form.getInneed().size() > 0) {
oracleParams.getName(), form.getTarget(), for (OracleConnectorParams oracleParams : form.getInneed()) {
oracleParams, OracleExtract); BasedTask task = new OracleExtractTask(
OracleExtractExecuter oee = new OracleExtractExecuter( oracleParams.getName(), form.getTarget(),
task); oracleParams, OracleExtract);
new Thread(oee, oracleParams.getName()).start(); OracleExtractExecuter oee = new OracleExtractExecuter(
} task);
} new Thread(oee, oracleParams.getName()).start();
} else }
req.setStatus(500); }
} } else
req.setStatus(500);
} }
/** }
* oracle汇总、抽取库的 查询
* /**
* @return * oracle汇总、抽取库的 查询
* @throws Exception *
*/ * @return
@RequestMapping(value = "/oracle/list", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public List<GatherOracleInfo> oracleList() throws Exception { @RequestMapping(value = "/oracle/list", method = RequestMethod.POST)
log.info("----------getOracleInfo-----------------------"); @ResponseBody
List<GatherOracleInfo> result = mySqlService.findAllOracle(); public List<GatherOracleInfo> oracleList() throws Exception {
return result; log.info("----------getOracleInfo-----------------------");
} List<GatherOracleInfo> result = mySqlService.findAllOracle();
return result;
/** }
* oracle汇总、抽取库的 删除
* /**
* @param req * oracle汇总、抽取库的 删除
* @param res *
* @param id * @param req
* @throws Exception * @param res
*/ * @param id
@RequestMapping(value = "/oracle/{id}/delete", method = RequestMethod.POST) * @throws Exception
public void oracleDelete(HttpServletRequest req, HttpServletResponse res, */
@PathVariable String id) throws Exception { @RequestMapping(value = "/oracle/{id}/delete", method = RequestMethod.POST)
log.info("----------deleteOracleInfo----------------"); public void oracleDelete(HttpServletRequest req, HttpServletResponse res,
Integer num = Integer.valueOf(id); @PathVariable String id) throws Exception {
Integer result = mySqlService.deleteOracle(num); log.info("----------deleteOracleInfo----------------");
log.info("oracleDelete : " + result); Integer num = Integer.valueOf(id);
res.setStatus(200); Integer result = mySqlService.deleteOracle(num);
} log.info("oracleDelete : " + result);
res.setStatus(200);
/** }
* oracle汇总、抽取库的 新增
* /**
* @param res * oracle汇总、抽取库的 新增
* @param req *
* @param id * @param res
* @param oracle * @param req
* @throws Exception * @param id
*/ * @param oracle
@RequestMapping(value = "/oracle/{id}/insert", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public void oracleInsert(HttpServletRequest res, HttpServletResponse req, @RequestMapping(value = "/oracle/{id}/insert", method = RequestMethod.POST)
@PathVariable String id, @RequestBody GatherOracleInfo oracle) @ResponseBody
throws Exception { public void oracleInsert(HttpServletRequest res, HttpServletResponse req,
log.info("----------insertOracleInfo----------------"); @PathVariable String id, @RequestBody GatherOracleInfo oracle)
oracle.setId(Integer.valueOf(id)); throws Exception {
mySqlService.insertOracle(oracle); log.info("----------insertOracleInfo----------------");
req.setStatus(200); oracle.setId(Integer.valueOf(id));
} mySqlService.insertOracle(oracle);
req.setStatus(200);
/** }
* oracle汇总、抽取库的 更新
* /**
* @param res * oracle汇总、抽取库的 更新
* @param req *
* @param id * @param res
* @param oracle * @param req
* @throws Exception * @param id
*/ * @param oracle
@RequestMapping(value = "/oracle/{id}/update", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public void oracleUpdate(HttpServletRequest res, HttpServletResponse req, @RequestMapping(value = "/oracle/{id}/update", method = RequestMethod.POST)
@PathVariable("id") String id, @RequestBody GatherOracleInfo oracle) @ResponseBody
throws Exception { public void oracleUpdate(HttpServletRequest res, HttpServletResponse req,
log.info("----------updateOracleInfo-----------------------"); @PathVariable("id") String id, @RequestBody GatherOracleInfo oracle)
log.info(oracle); throws Exception {
mySqlService.updateOracle(oracle); log.info("----------updateOracleInfo-----------------------");
req.setStatus(200); log.info(oracle);
} mySqlService.updateOracle(oracle);
req.setStatus(200);
/** }
* 迁移 数据 的查询
* /**
* @return * 迁移任务 的查询
* @throws Exception *
*/ * @return
@RequestMapping(value = "/task/transfer/list", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public Object taskTransferList() throws Exception { @RequestMapping(value = "/task/transfer/list", method = RequestMethod.POST)
log.debug("----------/task/transfer/list-----------------------"); @ResponseBody
List<DataInfoEntityMoveTmp> result = moveDataService.findAll(); public Object taskTransferList() throws Exception {
return result; log.debug("----------/task/transfer/list-----------------------");
} List<DataInfoEntityMoveTmp> result = moveDataService.findAll();
return result;
/** }
* 迁移完成后的 删除记录功能
* /**
* @param res * 迁移完成后的 删除记录功能
* @param req *
* @param id * @param res
* @return * @param req
* @throws Exception * @param id
*/ * @return
@RequestMapping(value = "/task/transfer/{id}/delete", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public Object taskTransferDelete(HttpServletRequest res, @RequestMapping(value = "/task/transfer/{id}/delete", method = RequestMethod.POST)
HttpServletResponse req, @PathVariable String id) throws Exception { @ResponseBody
log.info("----------/task/transfer/{id}/delete-----------------------"); public Object taskTransferDelete(HttpServletRequest res,
DataInfoEntityMoveTmp move = new DataInfoEntityMoveTmp(); HttpServletResponse req, @PathVariable String id) throws Exception {
move.setId(Integer.valueOf(id)); log.info("----------/task/transfer/{id}/delete-----------------------");
int result = moveDataService.delete(move); DataInfoEntityMoveTmp move = new DataInfoEntityMoveTmp();
req.setStatus(200); move.setId(Integer.valueOf(id));
return result; int result = moveDataService.delete(move);
} req.setStatus(200);
return result;
/** }
* 迁移数据完成后新增一条数据（暂时去掉，新增功能不在此）
* /**
* @param res * 迁移完成后的 删除记录功能
* @param req *
* @param move * @param res
* @return * @param req
* @throws Exception * @param id
*/ * @return
@RequestMapping(value = "/task/transfer/save", method = RequestMethod.POST) * @throws Exception
@ResponseBody */
public Object taskTransferSave(HttpServletRequest res, @RequestMapping(value = "/task/transfer/deletes", method = RequestMethod.POST)
HttpServletResponse req, @RequestBody DataInfoEntity move) @ResponseBody
throws Exception { public Object taskTransferDeletes(HttpServletRequest res,
log.info("---------/task/transfer/save-----------------------"); HttpServletResponse req, @RequestBody String[] ids) throws Exception {
// int result = dfs.save(move); log.info("---------/task/transfer/deletes----------------------");
req.setStatus(200); req.setStatus(200);
return 1; int result = 0;
} if (ids != null ) {
// String[] ids = idsString.split(",");
/** result = moveDataService.delete(ids);
* 地区和系统的 code 对应的名称 的 获取 }
* else {
* @return req.setStatus(500);
* @throws Exception }
*/ return result;
@RequestMapping(value = "/code/list", method = RequestMethod.POST) }
@ResponseBody
public Object findCodeList() throws Exception { /**
log.info("---------/findSystemCode-----------------------"); * 迁移数据完成后新增一条数据（暂时去掉，新增功能不在此）
Map<String, Object> result = codeService.findAll(); *
return result; * @param res
} * @param req
* @param move
/** * @return
* oracle 汇总的 日志 读取 * @throws Exception
* */
* @param name @RequestMapping(value = "/task/transfer/save", method = RequestMethod.POST)
* @param res @ResponseBody
* @param req public Object taskTransferSave(HttpServletRequest res,
* @return HttpServletResponse req, @RequestBody DataInfoEntity move)
* @throws Exception throws Exception {
*/ log.info("---------/task/transfer/save-----------------------");
@RequestMapping(value = "/oracle/extract/log", method = RequestMethod.POST) // int result = dfs.save(move);
@ResponseBody req.setStatus(200);
public Object getExtractLog(@RequestParam("rcName") String name, return 1;
HttpServletRequest res, HttpServletResponse req) throws Exception { }
log.info("---------/oracle/extract/log-------------------");
String result = logReadService.readLog(name); /**
// "查看相应日志" * 地区和系统的 code 对应的名称 的 获取
Map<String, String> log = new HashMap<String, String>(); *
log.put(name, result + "\r\n"); * @return
return log; * @throws Exception
} */
} @RequestMapping(value = "/code/list", method = RequestMethod.POST)
@ResponseBody
public Object findCodeList() throws Exception {
log.info("---------/findSystemCode-----------------------");
Map<String, Object> result = codeService.findAll();
return result;
}
/**
* oracle 汇总的 日志 读取
*
* @param name
* @param res
* @param req
* @return
* @throws Exception
*/
@RequestMapping(value = "/oracle/extract/log", method = RequestMethod.POST)
@ResponseBody
public Object getExtractLog(@RequestParam("rcName") String name,
HttpServletRequest res, HttpServletResponse req) throws Exception {
log.info("---------/oracle/extract/log-------------------");
String result = logReadService.readLog(name);
// "查看相应日志"
Map<String, String> log = new HashMap<String, String>();
log.put(name, result + "\r\n");
return log;
}
}

12
src/com/platform/form/PagerOptions.java
Unescape View File

@ -39,13 +39,13 @@ public class PagerOptions extends Page{
private Integer offset; // 查询偏移量：起始id private Integer offset; // 查询偏移量：起始id
private String keyQuery; private String keyQuery;//模糊查询字段
//模糊查询字段
private List<String> array;
private String volumeType; private List<String> array;//模糊查询字段数组
//冷热区字段
private String mark; private String volumeType;//冷热区字段
private String mark;//冷热区字段
public String getDataType() { public String getDataType() {
return dataType; return dataType;

10
src/com/platform/form/ScriptForm.java
Unescape View File

@ -5,10 +5,20 @@ import java.util.Properties;
import com.platform.entities.SqlFileInfoEntity; import com.platform.entities.SqlFileInfoEntity;
/**
* @author chen
* 脚本 表单接收
*/
public class ScriptForm { public class ScriptForm {
/**
* 脚本内容
*/
private String content; private String content;
/**
* 脚本实体内容
*/
private SqlFileInfoEntity item; private SqlFileInfoEntity item;
/** /**

2
src/com/platform/form/VolumeForm.java
Unescape View File

@ -29,9 +29,9 @@ public class VolumeForm {
/** 挂载点 */ /** 挂载点 */
private String path; private String path;
/** * exist，正常返回状态Started,Stopped,Created */
private boolean status; private boolean status;
/** * exist，正常返回状态Started,Stopped,Created */
private String type; private String type;
/** volume树形目录 */ /** volume树形目录 */

11
src/com/platform/form/volumeMoveForm.java
Unescape View File

@ -5,16 +5,25 @@ import java.util.List;
import com.platform.entities.DataInfoEntity; import com.platform.entities.DataInfoEntity;
import com.platform.entities.FolderNode; import com.platform.entities.FolderNode;
/**迁移数据--接收前端传值：volume名，目录，数据对象 /**迁移数据--接收前端传值
* @author chen * @author chen
* *
*/ */
public class volumeMoveForm { public class volumeMoveForm {
/**
* volume名
*/
private String name; private String name;
/**
* 目录
*/
private FolderNode selectNode; private FolderNode selectNode;
/**
* 数据对象
*/
private List<DataInfoEntity> selectItems; private List<DataInfoEntity> selectItems;
/** /**

6
src/com/platform/service/IMoveDataService.java
Unescape View File

@ -32,6 +32,12 @@ public interface IMoveDataService {
*/ */
public int delete(DataInfoEntityMoveTmp dataMove) throws Exception; public int delete(DataInfoEntityMoveTmp dataMove) throws Exception;
/** 删除多个
* @return
* @throws Exception
*/
public int delete(String... ids) throws Exception;
/** 迁移数据--更新 /** 迁移数据--更新
* @param data * @param data
* @return * @return

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;
} }
} }
} }

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

@ -76,9 +76,9 @@ public class CheckoutServiceImpl implements ICheckoutService {
List<CheckoutEntity> checks = new ArrayList<CheckoutEntity>(); List<CheckoutEntity> checks = new ArrayList<CheckoutEntity>();
List<PreDataInfo> result = preDataInfoDao.findAllCollect(); List<PreDataInfo> result = preDataInfoDao.findAllCollect();
DataInfoEntity data = new DataInfoEntity(); DataInfoEntity data = new DataInfoEntity();
Calendar c2 = Calendar.getInstance(); Calendar c2 = Calendar.getInstance();
// 时间设置为 半年前的时间 // 时间设置为 半年前的时间
c2.set(Calendar.MONTH, getMonBeforeHalfYear(c2.get(Calendar.MONTH))); this.getMonBeforeHalfYear(c2);
String time = DateForm.date2StringByDay(c2.getTime()); String time = DateForm.date2StringByDay(c2.getTime());
data.setCollectingTime(time); data.setCollectingTime(time);
// data.setCollectingTime(collectingTime); // data.setCollectingTime(collectingTime);
@ -124,7 +124,7 @@ public class CheckoutServiceImpl implements ICheckoutService {
CheckoutEntity cksql = new CheckoutEntity(); CheckoutEntity cksql = new CheckoutEntity();
Calendar c2 = Calendar.getInstance(); Calendar c2 = Calendar.getInstance();
// 时间设置为 半年前的时间 // 时间设置为 半年前的时间
c2.set(Calendar.MONTH, getMonBeforeHalfYear(c2.get(Calendar.MONTH))); this.getMonBeforeHalfYear(c2);
String time = DateForm.date2StringByDay(c2.getTime()); String time = DateForm.date2StringByDay(c2.getTime());
cksql.setCollectingTime(time); cksql.setCollectingTime(time);
cksql.setCityName(city); cksql.setCityName(city);
@ -451,12 +451,9 @@ public class CheckoutServiceImpl implements ICheckoutService {
* @param num * @param num
* @return * @return
*/ */
private int getMonBeforeHalfYear(int num){ private int getMonBeforeHalfYear(Calendar c2){
num -= Configs.dataBefore; c2.set(Calendar.DAY_OF_YEAR, c2.get(Calendar.DAY_OF_YEAR) -(Configs.dataBefore*30));
if (num <= 0) { return 1;
num = num + 12;
}
return num;
} }
private String isY(String str1, String str2) { private String isY(String str1, String str2) {

45
src/com/platform/service/impl/MoveDataServiceImpl.java
Unescape View File

@ -210,11 +210,14 @@ public class MoveDataServiceImpl implements IMoveDataService {
} }
} }
int result = 0;
//迁移失败 //迁移失败
if ("3".equals(dataMove.getCompleteStatus())) { if ("3".equals(dataMove.getCompleteStatus())) {
removeservice.deleteFolder(dataMove.getDstPath()); int code = removeservice.deleteFolder(dataMove.getDstPath());
if (code == -100) {
return result;
}
} }
int result = 0;
//是正则迁移时： //是正则迁移时：
if ("1".equals(dataMove.getCompleteStatus())) { if ("1".equals(dataMove.getCompleteStatus())) {
if(1 != removeservice.abortcopyFolder(dataMove.getDataPath(), makeDstPath(dataMove.getDstPath()))){ if(1 != removeservice.abortcopyFolder(dataMove.getDataPath(), makeDstPath(dataMove.getDstPath()))){
@ -263,4 +266,42 @@ public class MoveDataServiceImpl implements IMoveDataService {
} }
return dstPath; return dstPath;
} }
@Override
public int delete(String... ids) throws Exception {
List<DataInfoEntityMoveTmp> list = dataInfoMoveTmpDao.findAll();
List<DataInfoEntityMoveTmp> dellist = new ArrayList<DataInfoEntityMoveTmp>();
if (null != list) {
for (DataInfoEntityMoveTmp dataInfoEntityMoveTmp : list) {
for (String id : ids) {
if (Integer.valueOf(id) == dataInfoEntityMoveTmp.getId()) {
dellist.add(dataInfoEntityMoveTmp);
}
}
}
}
int result = 0;
for (DataInfoEntityMoveTmp dataMove : dellist) {
//迁移失败
if ("3".equals(dataMove.getCompleteStatus())) {
int code = removeservice.deleteFolder(dataMove.getDstPath());
if (code == -100) {
continue;
}
}
//是正则迁移时：
if ("1".equals(dataMove.getCompleteStatus())) {
if(1 != removeservice.abortcopyFolder(dataMove.getDataPath(), makeDstPath(dataMove.getDstPath()))){
try{
removeservice.deleteFolder(dataMove.getDstPath());
}catch(Exception e){
log.error(e);
}
}
}
result = dataInfoMoveTmpDao.remove(dataMove.getId());
}
return result;
}
} }

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;
// } // }
} }

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

@ -1,326 +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 "
+ Configs.KUBE_MASTER_ADDRESS String cmd = "kubectl --server "
+ " label --overwrite rc " + key + " status=1"; + Configs.KUBE_MASTER_ADDRESS
List<String> rList = Constant.ganymedSSH + " label --overwrite rc " + key + " status=1";
.execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); List<String> rList = Constant.ganymedSSH
for (String str : rList) .execCmdWaitAcquiescent(cmd);
sb.append(str).append("\n"); StringBuffer sb = new StringBuffer();
log.info("更新replicationController标签： " + key for (String str : rList)
+ "\t[标签更新为： 失败]"); sb.append(str).append("\n");
log.info(sb.toString()); log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 失败]");
String cmd2 = "kubectl --server " log.info(sb.toString());
+ Configs.KUBE_MASTER_ADDRESS
+ " annotate --overwrite rc " + key
+ " checkoutFlag=0"; String cmd2 = "kubectl --server "
List<String> rList2 = Constant.ganymedSSH + Configs.KUBE_MASTER_ADDRESS
.execCmdWaitAcquiescent(cmd2); + " annotate --overwrite rc " + key
StringBuffer sb2 = new StringBuffer(); + " checkoutFlag=0";
for (String str : rList2) List<String> rList2 = Constant.ganymedSSH
sb2.append(str).append("\n"); .execCmdWaitAcquiescent(cmd2);
log.info("更新replicationController标签： " + key StringBuffer sb2 = new StringBuffer();
+ "\t[标签更新为： 未校验]"); for (String str : rList2)
log.info(sb2.toString()); sb2.append(str).append("\n");
// 当前key标签对应的 数据服务的记录 log.info("更新replicationController标签： " + key
CheckoutEntity tmp = CacheOracleCheckoutEntity + "\t[标签更新为： 未校验]");
.getCheck(key); log.info(sb2.toString());
tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ZERO); // 当前key标签对应的 数据服务的记录
tmp.setPayResultLast(Constant.CHECKOUT_STATUS_ZERO); CheckoutEntity tmp = CacheOracleCheckoutEntity
tmp.setExecResultLast(Constant.CHECKOUT_STATUS_ZERO); .getCheck(key);
try { tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ZERO);
// 更新数据库 tmp.setPayResultLast(Constant.CHECKOUT_STATUS_ZERO);
this.updateDataInfo(tmp); tmp.setExecResultLast(Constant.CHECKOUT_STATUS_ZERO);
} catch (Exception e) { try {
log.error(e); // 更新数据库
} this.updateDataInfo(tmp);
} } catch (Exception e) {
} log.error(e);
} catch (Exception e1) { }
log.error(e1.getStackTrace()); }
} finally { }
try { } catch (Exception e1) {
Thread.sleep(1000 * 60); log.error(e1.getStackTrace());
} catch (InterruptedException e) { } finally {
log.error(e); try {
} Thread.sleep(1000 * 60);
} } catch (InterruptedException e) {
} log.error(e);
} }
}
/** }
* 查找 pod 节点 }
*
* @param taskName /**
* @return * 查找 pod 节点
*/ *
private Pod filterPod(String taskName) { * @param taskName
Pod pod = null; * @return
ReplicationController replicationController = client */
.getReplicationController(taskName); private Pod filterPod(String taskName) {
if (null != replicationController) { Pod pod = null;
List<Pod> filterPods = client ReplicationController replicationController = client
.getPodsForApplicaList(replicationController); .getReplicationController(taskName);
if (filterPods != null && filterPods.size() > 0) { if (null != replicationController) {
pod = filterPods.get(0); List<Pod> filterPods = client
} .getPodsForApplicaList(replicationController);
} if (filterPods != null && filterPods.size() > 0) {
return pod; pod = filterPods.get(0);
} }
}
/** return pod;
* 尝试连接 oracle服务 }
*
* @param tmpPod /**
* @param key * 尝试连接 oracle服务
*/ *
private void connectOracle(Pod pod, String key) throws Exception { * @param tmpPod
if (pod != null) { * @param key
String ip = client.getPodHostIp(pod); */
int port = client.getPodContainerport(pod); private void connectOracle(Pod pod, String key) throws Exception {
if (ip != null && port != 0) { if (pod != null) {
String url = "jdbc:oracle:thin:@" + ip + ":" + port + ":" String ip = client.getPodHostIp(pod);
+ Configs.ORACLE_ORCL; int port = client.getPodContainerport(pod);
boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败 if (ip != null && port != 0) {
Configs.ORACLE_USER, Configs.ORACLE_PSW); String url = "jdbc:oracle:thin:@" + ip + ":" + port + ":"
log.info("url:" + url + ",user:" + Configs.ORACLE_USER + Configs.ORACLE_ORCL;
+ ",password:" + Configs.ORACLE_PSW); boolean flag = OracleConnector.canConnect(url, // 连接结果返回参数，true标示连接成功，false标示连接失败
String message = "失败"; Configs.ORACLE_USER, Configs.ORACLE_PSW);
log.info("url:" + url + ",user:" + Configs.ORACLE_USER
String cmd3 = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS + ",password:" + Configs.ORACLE_PSW);
+ " annotate --overwrite rc " + key + " checkoutFlag=0"; String message = "失败";
List<String> rList3 = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd3);
StringBuffer sb3 = new StringBuffer(); String cmd3 = "kubectl --server " + Configs.KUBE_MASTER_ADDRESS
for (String str : rList3) + " annotate --overwrite rc " + key + " checkoutFlag=0";
sb3.append(str).append("\n");
log.info(sb3.toString()); List<String> rList3 = Constant.ganymedSSH
log.info("更新replicationController标签： " + key + "\t[标签更新为：未校验]"); .execCmdWaitAcquiescent(cmd3);
StringBuffer sb3 = new StringBuffer();
if (flag) { for (String str : rList3)
String cmd = "kubectl --server " sb3.append(str).append("\n");
+ Configs.KUBE_MASTER_ADDRESS log.info(sb3.toString());
+ " label --overwrite rc " + key + " status=2"; log.info("更新replicationController标签： " + key + "\t[标签更新为：未校验]");
// 设置服务为 成功
List<String> rList = Constant.ganymedSSH if (flag) {
.execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); String cmd = "kubectl --server "
for (String str : rList) + Configs.KUBE_MASTER_ADDRESS
sb.append(str).append("\n"); + " label --overwrite rc " + key + " status=2";
log.info(sb.toString());
message = "成功"; // 设置服务为 成功
log.info("更新replicationController标签： " + key List<String> rList = Constant.ganymedSSH
+ "\t[标签更新为： 成功]"); .execCmdWaitAcquiescent(cmd);
// 校验标签 StringBuffer sb = new StringBuffer();
String cmd2 = "kubectl --server " for (String str : rList)
+ Configs.KUBE_MASTER_ADDRESS sb.append(str).append("\n");
+ " annotate --overwrite rc " + key log.info(sb.toString());
+ " checkoutFlag=2"; message = "成功";
List<String> rList2 = Constant.ganymedSSH log.info("更新replicationController标签： " + key
.execCmdWaitAcquiescent(cmd2); + "\t[标签更新为： 成功]");
StringBuffer sb2 = new StringBuffer(); // 校验标签
for (String str : rList2)
sb2.append(str).append("\n"); String cmd2 = "kubectl --server "
log.info(sb2.toString()); + Configs.KUBE_MASTER_ADDRESS
log.info("更新replicationController标签： " + key + " annotate --overwrite rc " + key
+ "\t[标签更新为： 校验中]"); + " checkoutFlag=2";
// 获得当前 服务对应的 数据 List<String> rList2 = Constant.ganymedSSH
CheckoutEntity tmp = CacheOracleCheckoutEntity .execCmdWaitAcquiescent(cmd2);
.getCheck(key); StringBuffer sb2 = new StringBuffer();
if (null != tmp) { for (String str : rList2)
CacheOracleCheckoutEntity.putExtract(key, tmp); sb2.append(str).append("\n");
} log.info(sb2.toString());
// 查询 对应的 2 个标准表 log.info("更新replicationController标签： " + key
OracleConnectorParams oc = new OracleConnectorParams(); + "\t[标签更新为： 校验中]");
String logName = tmp.getAreaCode().toLowerCase() + "_" // 获得当前 服务对应的 数据
+ tmp.getSysCode() + "_" + tmp.getDataVersion(); CheckoutEntity tmp = CacheOracleCheckoutEntity
oc.setName(logName); .getCheck(key);
try { if (null != tmp) {
Connection conn = OracleConnector.connectionBuilder( CacheOracleCheckoutEntity.putExtract(key, tmp);
url, Configs.ORACLE_USER, Configs.ORACLE_PSW, }
oc); // 查询 对应的 2 个标准表
// 支付--校验 OracleConnectorParams oc = new OracleConnectorParams();
if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp String logName = tmp.getAreaCode().toLowerCase() + "_"
.getPayResultLast())) { + tmp.getSysCode() + "_" + tmp.getDataVersion();
String paySql = "select * from dba_tables where owner = '" oc.setName(logName);
+ Configs.COLLECT_STANDARD_TABLE_USER try {
.toUpperCase() Connection conn = OracleConnector.connectionBuilder(
+ "' and table_name = '" url, Configs.ORACLE_USER, Configs.ORACLE_PSW,
+ Configs.COLLECT_PAY_TABLE.toUpperCase() oc);
+ "'"; // 支付--校验
if (OracleConnector if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp
.execUpdateOracleSQL( .getPayResultLast())) {
conn, String paySql = "select * from dba_tables where owner = '"
paySql, + Configs.COLLECT_STANDARD_TABLE_USER
FileOperateHelper .toUpperCase()
.addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION) + "' and table_name = '"
+ logName + "jy.log")) { + Configs.COLLECT_PAY_TABLE.toUpperCase()
tmp.setPayResultLast(Constant.CHECKOUT_STATUS_THREE); + "'";
} else { if (OracleConnector
tmp.setPayResultLast(Constant.CHECKOUT_STATUS_FOUR); .execUpdateOracleSQL(
} conn,
} paySql,
// 可执行-- 校验 FileOperateHelper
if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp .addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION)
.getExecResultLast())) { + logName + "jy.log")) {
String execSql = "select * from dba_tables where owner = '" tmp.setPayResultLast(Constant.CHECKOUT_STATUS_THREE);
+ Configs.COLLECT_STANDARD_TABLE_USER } else {
.toUpperCase() tmp.setPayResultLast(Constant.CHECKOUT_STATUS_FOUR);
+ "' and table_name = '" }
+ Configs.COLLECT_EXEC_TABLE.toUpperCase() }
+ "'"; // 可执行-- 校验
if (OracleConnector if (!Constant.CHECKOUT_STATUS_ONE.equals(tmp
.execUpdateOracleSQL( .getExecResultLast())) {
conn, String execSql = "select * from dba_tables where owner = '"
execSql, + Configs.COLLECT_STANDARD_TABLE_USER
FileOperateHelper .toUpperCase()
.addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION) + "' and table_name = '"
+ logName + "jy.log")) { + Configs.COLLECT_EXEC_TABLE.toUpperCase()
tmp.setExecResultLast(Constant.CHECKOUT_STATUS_THREE); + "'";
} else { if (OracleConnector
tmp.setExecResultLast(Constant.CHECKOUT_STATUS_FOUR); .execUpdateOracleSQL(
} conn,
} execSql,
tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ONE); FileOperateHelper
// 更新数据库data_info .addLastSeparator(Configs.EXTRACT_STANDARD_LOG_LOCALTION)
updateDataInfo(tmp); + logName + "jy.log")) {
// 将 CacheOracleCheckoutEntity checkMap中的 数据 tmp.setExecResultLast(Constant.CHECKOUT_STATUS_THREE);
// 放入extractMap中 } else {
/* CacheOracleCheckoutEntity.putExtract(key, */CacheOracleCheckoutEntity tmp.setExecResultLast(Constant.CHECKOUT_STATUS_FOUR);
.checkRemove(key)/* ) */; }
// 更新kuber状态 }
cmd2 = "kubectl --server " tmp.setCheckoutFlag(Constant.CHECKOUTFLAG_ONE);
+ Configs.KUBE_MASTER_ADDRESS // 更新数据库data_info
+ " annotate --overwrite rc " + key updateDataInfo(tmp);
+ " checkoutFlag=1"; // 将 CacheOracleCheckoutEntity checkMap中的 数据
// client.updateOrAddReplicasLabelById(taskNSyame, // 放入extractMap中
// "status", "2"); /* CacheOracleCheckoutEntity.putExtract(key, */CacheOracleCheckoutEntity
rList = Constant.ganymedSSH .checkRemove(key)/* ) */;
.execCmdWaitAcquiescent(cmd2); // 更新kuber状态
sb = new StringBuffer();
for (String str : rList) cmd2 = "kubectl --server "
sb.append(str).append("\n"); + Configs.KUBE_MASTER_ADDRESS
log.info(sb.toString()); + " annotate --overwrite rc " + key
message = "成功"; + " checkoutFlag=1";
log.info("更新replicationController标签： " + key // client.updateOrAddReplicasLabelById(taskNSyame,
+ "\t[标签更新为： 已校验]"); // "status", "2");
} catch (Exception e) { rList = Constant.ganymedSSH
log.error(e); .execCmdWaitAcquiescent(cmd2);
} sb = new StringBuffer();
// 成功 就 清除 CacheOracleCheckoutEntity 中 的该条记录 for (String str : rList)
CacheOracleCheckoutEntity.checkRemove(key); sb.append(str).append("\n");
} log.info(sb.toString());
log.info("连接到数据库服务： " + key + "\t[连接结果： " + message + "]"); message = "成功";
} log.info("更新replicationController标签： " + key
+ "\t[标签更新为： 已校验]");
} } catch (Exception e) {
} log.error(e);
}
/** // 成功 就 清除 CacheOracleCheckoutEntity 中 的该条记录
* 更新 dataInfo表 CacheOracleCheckoutEntity.checkRemove(key);
* }
* @param checkoutEntity log.info("连接到数据库服务： " + key + "\t[连接结果： " + message + "]");
* @throws Exception }
*/
private void updateDataInfo(CheckoutEntity checkoutEntity) throws Exception { }
DataInfoEntity data = new DataInfoEntity(); }
// 状态改为 正则校验 Constant.CHECKOUTFLAG_TWO
data.setId(checkoutEntity.getDataId()); /**
data.setPayResultLast(checkoutEntity.getPayResultLast()); * 更新 dataInfo表
data.setExecResultLast(checkoutEntity.getExecResultLast()); *
data.setCheckoutFlag(checkoutEntity.getCheckoutFlag()); * @param checkoutEntity
dataInfoDao.update(data); * @throws Exception
DataInfoEntity tmpdata = dataInfoDao.findById(checkoutEntity */
.getDataId()); private void updateDataInfo(CheckoutEntity checkoutEntity) throws Exception {
data.setId(tmpdata.getSrcId()); DataInfoEntity data = new DataInfoEntity();
dataInfoDao.update(data); // 状态改为 正则校验 Constant.CHECKOUTFLAG_TWO
} data.setId(checkoutEntity.getDataId());
} data.setPayResultLast(checkoutEntity.getPayResultLast());
data.setExecResultLast(checkoutEntity.getExecResultLast());
data.setCheckoutFlag(checkoutEntity.getCheckoutFlag());
dataInfoDao.update(data);
DataInfoEntity tmpdata = dataInfoDao.findById(checkoutEntity
.getDataId());
data.setId(tmpdata.getSrcId());
dataInfoDao.update(data);
}
}

15
src/com/platform/service/thread/ThreadExtractStandardSqlServer.java
Unescape View File

@ -8,6 +8,7 @@ import java.io.IOException;
import java.io.InputStreamReader; import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException; import java.io.UnsupportedEncodingException;
import java.sql.Connection; import java.sql.Connection;
import java.sql.SQLException;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@ -154,10 +155,11 @@ public class ThreadExtractStandardSqlServer extends Thread{
// TODO Auto-generated catch block // TODO Auto-generated catch block
e.printStackTrace(); e.printStackTrace();
} }
Connection conn = null;
try { try {
File execSql = new File(execFilePath); conn = OracleConnector.connectionBuilder("jdbc:oracle:thin:@" + oracleConnect.getIp() + ":" + oracleConnect.getPort() + ":"
Connection conn = OracleConnector.connectionBuilder("jdbc:oracle:thin:@" + oracleConnect.getIp() + ":" + oracleConnect.getPort() + ":"
+ oracleConnect.getDatabaseName(), oracleConnect.getUser(), oracleConnect.getPassword(), null); + oracleConnect.getDatabaseName(), oracleConnect.getUser(), oracleConnect.getPassword(), null);
File execSql = new File(execFilePath);
// 创建表空间 创建 抽取标准表的 用户并授权 // 创建表空间 创建 抽取标准表的 用户并授权
oracleExtract.createTableSpace(conn, collectOracle , oracleConnect); // oracleExtract.createTableSpace(conn, collectOracle , oracleConnect); //
oracleExtract.createOnlyUser(conn, collectOracle, oracleConnect);// oracleExtract.createOnlyUser(conn, collectOracle, oracleConnect);//
@ -199,8 +201,17 @@ public class ThreadExtractStandardSqlServer extends Thread{
// TODO Auto-generated catch block // TODO Auto-generated catch block
e.printStackTrace(); e.printStackTrace();
} }
finally{
try {
conn.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
} }
//删除保存过的id //删除保存过的id
CacheSetCantDelete.removeStandardId(String.valueOf(element.getDataId())); CacheSetCantDelete.removeStandardId(String.valueOf(element.getDataId()));
// sql日志记录时间： // sql日志记录时间：

259
src/com/platform/utils/Configs.java
Unescape View File

@ -1,129 +1,130 @@
package com.platform.utils; package com.platform.utils;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
/** 配置启动 变量 /** 配置启动 变量
* @author chen * @author chen
* *
*/ */
public class Configs { public class Configs {
/** 全局自定义异常--编码 */ /** 全局自定义异常--编码 */
public static final String GLOBAL_EXP_CUSTOM = "3001001001"; public static final String GLOBAL_EXP_CUSTOM = "3001001001";
/** 全局非自定义异常--编码 */ /** 全局非自定义异常--编码 */
public static final String GLOBAL_EXP_NOT_CUSTOM = "3001001002"; public static final String GLOBAL_EXP_NOT_CUSTOM = "3001001002";
public static final String CONFIG_LOCALTION = "WebContent/WEB-INF/config/config.properties"; public static final String CONFIG_LOCALTION = "WebContent/WEB-INF/config/config.properties";
public static final Logger CONSOLE_LOGGER = Logger.getLogger("console"); public static final Logger CONSOLE_LOGGER = Logger.getLogger("console");
public static final Logger DAILY_ROLLING_LOGGER = Logger public static final Logger DAILY_ROLLING_LOGGER = Logger
.getLogger("dailyRollingFile"); .getLogger("dailyRollingFile");
public static final Logger DAILY_LOGGER = Logger.getLogger("railyFile"); public static final Logger DAILY_LOGGER = Logger.getLogger("railyFile");
public static final Logger LOGGER = Logger.getLogger(Configs.class); public static final Logger LOGGER = Logger.getLogger(Configs.class);
public static String KUBE_MASTER_URL = "http://192.168.0.110:8080/"; // kubernetes集群的maser public static String KUBE_MASTER_URL = "http://192.168.0.110:8080/"; // kubernetes集群的maser
// URl // URl
public static String KUBE_MASTER_ADDRESS="127.0.0.1:8080"; public static String KUBE_MASTER_ADDRESS="127.0.0.1:8080";
public static int ORACLE_DEFAULT_PORT = 1521; // oracle的默认端口号
public static int ORACLE_DEFAULT_PORT = 1521; // oracle的默认端口号
public static String COLLECT_USER_NAME = "system"; //采集统一的登入用户名
public static String COLLECT_USER_NAME = "system"; //采集统一的登入用户名
public static String COLLECT_PASSWORD = "oracle"; //采集统一的登入密码
public static String COLLECT_PASSWORD = "oracle"; //采集统一的登入密码
public static String COLLECT_SERVICE_NAME = "orcl"; //采集库统一的服务名
public static String COLLECT_SERVICE_NAME = "orcl"; //采集库统一的服务名
public static String GATHER_PORT ="1521"; //汇总库的端口号
public static String GATHER_PORT ="1521"; //汇总库的端口号
public static String GATHER_USER_NAME = "system"; //汇总库的登入用户名
public static String GATHER_USER_NAME = "system"; //汇总库的登入用户名
public static String GATHER_USER_PASSWORD = "1"; //汇总库的登入密码
public static String GATHER_USER_PASSWORD = "1"; //汇总库的登入密码
/** 抽取标准表 的 标准表所在 用户 */
public static String GATHER_STANDARD_USER_NAME = "u_bzbjy"; /** 抽取标准表 的 标准表所在 用户 */
public static String GATHER_STANDARD_USER_NAME = "u_bzbjy";
/** 可执行表名 */
public static String GATHER_STANDARD_EXEC_TABLE_NAME = "kzxzb"; /** 可执行表名 */
public static String GATHER_STANDARD_EXEC_TABLE_NAME = "kzxzb";
/** 可支付表名 */
public static String GATHER_STANDARD_PAY_TABLE_NAME = "zfxxb"; /** 可支付表名 */
public static String GATHER_STANDARD_PAY_TABLE_NAME = "zfxxb";
public static String GATHER_SERVICE_NAME = "orcl"; //汇总库的服务名
public static String GATHER_SERVICE_NAME = "orcl"; //汇总库的服务名
public static String TABLE_SUFFIX = "_20152016"; //汇总库汇总表的后缀名
public static String TABLE_SUFFIX = "_20152016"; //汇总库汇总表的后缀名
public static String EXTRACT_LOG_LOCALTION = "D:\\log"; //数据汇总日志保存位置
public static String EXTRACT_LOG_LOCALTION = "D:\\log"; //数据汇总日志保存位置
public static String EXTRACT_STANDARD_LOG_LOCALTION = "D:\\log2"; //数据抽取日志保存位置
public static String EXTRACT_STANDARD_LOG_LOCALTION = "D:\\log2"; //数据抽取日志保存位置
public static String GATHER_TABLESPACE_NAME=""; //表空间名
public static String GATHER_TABLESPACE_NAME=""; //表空间名
public static String GATHER_TABLESPACE_PATH=""; //表空间路径
public static String GATHER_TABLESPACE_PATH=""; //表空间路径
public static String GATHER_TABLE_PASSWORD="1"; //登入密码
public static String GATHER_TABLE_PASSWORD="1"; //登入密码
/** excel上传路径 */
public static String FILE_UPLOAD_PATH=""; /** excel上传路径 */
public static String FILE_UPLOAD_PATH="";
/** excel下载路经 */
public static String FILE_DOWNLOAD_PATH=""; /** excel下载路经 */
public static String FILE_DOWNLOAD_PATH="";
/** 包下载路径 */
public static String PACKAGE_DOWNLOAD_PATH=""; /** 包下载路径 */
public static String PACKAGE_DOWNLOAD_PATH="";
/** 下载包名 */
public static String PACKAGE_NAME=""; /** 下载包名 */
public static String PACKAGE_NAME="";
/**
* 最新脚本的位置 /**
*/ * 最新脚本的位置
public static String SQL_SCRIPT_PATH_LAST="D:\\test\\sql_script_last\\"; */
/** 归档脚本的位置 */ public static String SQL_SCRIPT_PATH_LAST="D:\\test\\sql_script_last\\";
public static String SQL_SCRIPT_PATH_STANDARD="D:\\test\\sql_script_standard\\"; /** 归档脚本的位置 */
public static String SQL_SCRIPT_PATH_STANDARD="D:\\test\\sql_script_standard\\";
/**
* 多少 个月前的数据 /**
*/ * 多少 个月前的数据
public static int dataBefore = 6; */
public static int dataBefore = 6;
/**
* excel导入一次批量导入5个 /**
*/ * excel导入一次批量导入5个
public static int NUM_ONE_IMPORT_EXCEL = 5; */
public static int NUM_ONE_IMPORT_EXCEL = 5;
/**
* docker启动的 oracle服务 的 实例名称 /**
*/ * docker启动的 oracle服务 的 实例名称
public static String ORACLE_ORCL = "orcl"; */
public static String ORACLE_ORCL = "orcl";
/**
* docker启动的 oracle服务 的 实例登陆用户名 /**
*/ * docker启动的 oracle服务 的 实例登陆用户名
public static String ORACLE_USER = "system"; */
public static String ORACLE_USER = "system";
/**
* docker启动的 oracle服务 的 实例登陆密码 /**
*/ * docker启动的 oracle服务 的 实例登陆密码
public static String ORACLE_PSW = "oracle"; */
public static String ORACLE_PSW = "oracle";
/**
* 采集库中的 标准表所在 的 用户 /**
*/ * 采集库中的 标准表所在 的 用户
public static String COLLECT_STANDARD_TABLE_USER = "u_bzbjy"; */
public static String COLLECT_STANDARD_TABLE_USER = "u_bzbjy";
/**
* 可执行的 标准表的 名 /**
*/ * 可执行的 标准表的 名
public static String COLLECT_EXEC_TABLE = "kzxzb"; */
public static String COLLECT_EXEC_TABLE = "kzxzb";
/**
* 支付的 标准表的 名 /**
*/ * 支付的 标准表的 名
public static String COLLECT_PAY_TABLE = "zfxxb"; */
public static String COLLECT_PAY_TABLE = "zfxxb";
}
}

3
src/com/platform/utils/GanymedSSH.java
Unescape View File

@ -114,6 +114,9 @@ public class GanymedSSH {
Session sess = null; Session sess = null;
try { try {
// conn = getOpenedConnection(host, username, password, port); // conn = getOpenedConnection(host, username, password, port);
if (null != conn) {
// conn=getOpenedConnection(Configs., username, password, port)
}
sess = conn.openSession(); sess = conn.openSession();
// 执锟斤拷cmd // 执锟斤拷cmd
sess.execCommand(cmd); sess.execCommand(cmd);

Write Preview
Loading…
Cancel
Save