diff --git a/canal-admin/canal-admin-server/pom.xml b/canal-admin/canal-admin-server/pom.xml index 65a2da76..3e031738 100644 --- a/canal-admin/canal-admin-server/pom.xml +++ b/canal-admin/canal-admin-server/pom.xml @@ -20,6 +20,10 @@ org.springframework.boot spring-boot-starter-jdbc + + org.springframework.boot + spring-boot-starter-test + mysql @@ -33,6 +37,21 @@ com.github.ben-manes.caffeine caffeine + + + io.ebean + ebean + 11.41.1 + + + + commons-beanutils + commons-beanutils + + + commons-lang + commons-lang + @@ -87,6 +106,17 @@ + + io.repaint.maven + tiles-maven-plugin + 2.12 + true + + + io.ebean.tile:enhancement:11.41.1 + + + diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/CanalAdminApplication.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/CanalAdminApplication.java index 5de3dfa3..4ff5f70c 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/CanalAdminApplication.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/CanalAdminApplication.java @@ -18,4 +18,9 @@ public class CanalAdminApplication { application.setBannerMode(Banner.Mode.OFF); application.run(args); } + + // public static void main(String[] args) throws Exception { + // UserAgent userAgent = new UserAgent(); + //// HelloAgent helloAgent = new HelloAgent + // } } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/Config.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/Config.java index 7f856a9a..1a25bc13 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/Config.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/Config.java @@ -1,15 +1,23 @@ package com.alibaba.otter.canal.admin.config; -import org.apache.commons.dbutils.QueryRunner; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - import javax.sql.DataSource; +import io.ebean.EbeanServer; +import org.apache.commons.dbutils.QueryRunner; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + @Configuration public class Config { + @Bean public QueryRunner queryRunner(DataSource dataSource) { return new QueryRunner(dataSource); } + +// @Bean("ebeanServer") +// public FactoryBean ebeanServer(DataSource dataSource) { +// return new EbeanServerFactory(dataSource); +// } } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/EbeanConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/EbeanConfig.java new file mode 100644 index 00000000..aec83c04 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/EbeanConfig.java @@ -0,0 +1,34 @@ +package com.alibaba.otter.canal.admin.config; + +import java.util.ArrayList; +import java.util.List; + +import javax.sql.DataSource; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import io.ebean.EbeanServer; +import io.ebean.EbeanServerFactory; +import io.ebean.config.ServerConfig; +import io.ebean.config.UnderscoreNamingConvention; + +@Configuration +public class EbeanConfig { + + @Bean("ebeanServer") + public EbeanServer ebeanServer(DataSource dataSource) { + ServerConfig serverConfig = new ServerConfig(); + serverConfig.setDefaultServer(true); + serverConfig.setNamingConvention(new UnderscoreNamingConvention()); + List packages = new ArrayList<>(); + packages.add("com.alibaba.otter.canal.admin.model"); + serverConfig.setPackages(packages); + serverConfig.setName("ebeanServer"); + serverConfig.setDataSource(dataSource); + serverConfig.setDatabaseSequenceBatchSize(1); + serverConfig.setDdlGenerate(false); + serverConfig.setDdlRun(false); + return EbeanServerFactory.create(serverConfig); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java index 9133c24d..25fe2e9b 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java @@ -43,37 +43,37 @@ public class WebConfig implements WebMvcConfigurer { } }).addPathPatterns("/api/**"); - registry.addInterceptor(new HandlerInterceptor() { - - @Override - public boolean preHandle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, - Object o) throws Exception { - String token = httpServletRequest.getHeader("X-Token"); - boolean valid = false; - if (token != null) { - User user = UserController.loginUsers.getIfPresent(token); - if (user != null) { - valid = true; - } - } - if (!valid) { - BaseModel baseModel = BaseModel.getInstance(null); - baseModel.setCode(50014); - baseModel.setMessage("Expired token"); - ObjectMapper mapper = new ObjectMapper(); - String json = mapper.writeValueAsString(baseModel); - try { - httpServletResponse.setContentType("application/json;charset=UTF-8"); - PrintWriter out = httpServletResponse.getWriter(); - out.print(json); - } catch (Exception e) { - e.printStackTrace(); - } - return false; - } - - return true; - } - }).addPathPatterns("/api/**").excludePathPatterns("/api/**/user/**"); +// registry.addInterceptor(new HandlerInterceptor() { +// +// @Override +// public boolean preHandle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, +// Object o) throws Exception { +// String token = httpServletRequest.getHeader("X-Token"); +// boolean valid = false; +// if (token != null) { +// User user = UserController.loginUsers.getIfPresent(token); +// if (user != null) { +// valid = true; +// } +// } +// if (!valid) { +// BaseModel baseModel = BaseModel.getInstance(null); +// baseModel.setCode(50014); +// baseModel.setMessage("Expired token"); +// ObjectMapper mapper = new ObjectMapper(); +// String json = mapper.writeValueAsString(baseModel); +// try { +// httpServletResponse.setContentType("application/json;charset=UTF-8"); +// PrintWriter out = httpServletResponse.getWriter(); +// out.print(json); +// } catch (Exception e) { +// e.printStackTrace(); +// } +// return false; +// } +// +// return true; +// } +// }).addPathPatterns("/api/**").excludePathPatterns("/api/**/user/**"); } } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/NodeServerController.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/NodeServerController.java new file mode 100644 index 00000000..f29d1a16 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/NodeServerController.java @@ -0,0 +1,45 @@ +package com.alibaba.otter.canal.admin.controller; + +import java.util.List; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.*; + +import com.alibaba.otter.canal.admin.model.BaseModel; +import com.alibaba.otter.canal.admin.model.NodeServer; +import com.alibaba.otter.canal.admin.service.NodeServerService; + +@RestController +@RequestMapping("/api/{env}") +public class NodeServerController { + + @Autowired + NodeServerService nodeServerService; + + @GetMapping(value = "/nodeServers") + public BaseModel> nodeServers(NodeServer nodeServer, @PathVariable String env) { + return BaseModel.getInstance(nodeServerService.findList(nodeServer)); + } + + @DeleteMapping(value = "/nodeServer/{id}") + public BaseModel delete(@PathVariable Long id, @PathVariable String env) { + nodeServerService.delete(id); + return BaseModel.getInstance("success"); + } + + @PostMapping(value = "/nodeServer") + public BaseModel save(@RequestBody NodeServer nodeServer, @PathVariable String env) { + nodeServerService.save(nodeServer); + return BaseModel.getInstance("success"); + } + + @GetMapping(value = "/nodeServer/{id}") + public BaseModel detail(@PathVariable Long id, @PathVariable String env) { + return BaseModel.getInstance(nodeServerService.detail(id)); + } + + @GetMapping(value = "/nodeServer/status") + public BaseModel status(@RequestParam String ip, @RequestParam Integer port, @PathVariable String env) { + return BaseModel.getInstance(nodeServerService.remoteNodeStatus(ip,port)); + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java deleted file mode 100644 index 71f177c0..00000000 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/dao/CanalConfigDao.java +++ /dev/null @@ -1,35 +0,0 @@ -package com.alibaba.otter.canal.admin.dao; - -import com.alibaba.otter.canal.admin.model.CanalConfig; -import org.apache.commons.dbutils.QueryRunner; -import org.apache.commons.dbutils.handlers.BeanHandler; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Repository; - -import java.sql.SQLException; -import java.util.Date; - -@Repository -public class CanalConfigDao { - - @Autowired - private QueryRunner runner; - - public CanalConfig findById(Long id) { - try { - String sql = "select id,name,content,modified_time as modifiedTime" + " from canal_config where id=?"; - return runner.query(sql, new BeanHandler<>(CanalConfig.class), id); - } catch (SQLException e) { - throw new RuntimeException(e.getMessage(), e); - } - } - - public int updateContent(CanalConfig canalConfig){ - try { - String sql = "update canal_config set content=?, modified_time=? where id=?"; - return runner.update(sql, canalConfig.getContent(), new Date(), canalConfig.getId()); - } catch (SQLException e) { - throw new RuntimeException(e.getMessage(), e); - } - } -} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/handler/CustomExceptionHandler.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/handler/CustomExceptionHandler.java new file mode 100644 index 00000000..bcb70a5c --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/handler/CustomExceptionHandler.java @@ -0,0 +1,32 @@ +package com.alibaba.otter.canal.admin.handler; + +import com.alibaba.otter.canal.admin.common.exception.ServiceException; +import com.alibaba.otter.canal.admin.model.BaseModel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.ControllerAdvice; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.ResponseBody; +import org.springframework.web.bind.annotation.ResponseStatus; + +@ControllerAdvice(annotations = ResponseBody.class) +public class CustomExceptionHandler { + + private static Logger logger = LoggerFactory.getLogger(CustomExceptionHandler.class); + + @ResponseBody + @ResponseStatus(HttpStatus.OK) + @ExceptionHandler(value = Exception.class) + public BaseModel commonExceptionHandle(Exception e) { + if (e instanceof ServiceException) { + logger.error(e.getMessage()); + } else { + logger.error(e.getMessage(), e); + } + BaseModel res = new BaseModel(); + res.setCode(50000); + res.setMessage(e.getMessage()); + return res; + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/jmx/CanalServerMXBean.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/jmx/CanalServerMXBean.java new file mode 100644 index 00000000..c3a7089a --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/jmx/CanalServerMXBean.java @@ -0,0 +1,22 @@ +package com.alibaba.otter.canal.admin.jmx; + +public interface CanalServerMXBean { + + int getStatus(); + + boolean start(); + + boolean stop(); + + boolean restart(); + + boolean exit(); + + boolean startInstance(String destination); + + boolean stopInstance(String destination); + + boolean reloadInstance(String destination); + + String getRunningInstances(); +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/jmx/JMXConnection.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/jmx/JMXConnection.java new file mode 100644 index 00000000..6e4140ed --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/jmx/JMXConnection.java @@ -0,0 +1,115 @@ +package com.alibaba.otter.canal.admin.jmx; + +import com.alibaba.otter.canal.admin.common.exception.ServiceException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.management.*; +import javax.management.remote.JMXConnector; +import javax.management.remote.JMXConnectorFactory; +import javax.management.remote.JMXServiceURL; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.SocketTimeoutException; +import java.util.concurrent.*; + +public class JMXConnection { + + private Logger logger = LoggerFactory.getLogger(JMXConnection.class); + + private String ip; + private Integer port; + private JMXConnector jmxc; + private CanalServerMXBean canalServerMXBean; + + public JMXConnection(String ip, Integer port){ + this.ip = ip; + this.port = port; + } + + public void connect() { + try { + JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://" + ip + ":" + port + "/jmxrmi"); + // jmxc = JMXConnectorFactory.connect(url, null); + jmxc = connectWithTimeout(url, 3, TimeUnit.SECONDS); + MBeanServerConnection mBeanServerConnection = jmxc.getMBeanServerConnection(); + ObjectName name = new ObjectName("CanalServerAgent:type=CanalServerStatus"); + mBeanServerConnection.addNotificationListener(name, (notification, handback) -> { + }, null, null); + canalServerMXBean = MBeanServerInvocationHandler + .newProxyInstance(mBeanServerConnection, name, CanalServerMXBean.class, false); + } catch (Exception e) { + throw new ServiceException(e.getMessage(), e); + } + } + + private static JMXConnector connectWithTimeout(final JMXServiceURL url, long timeout, + TimeUnit unit) throws IOException { + final BlockingQueue mailbox = new ArrayBlockingQueue<>(1); + ExecutorService executor = Executors.newSingleThreadExecutor(daemonThreadFactory); + executor.submit(() -> { + try { + JMXConnector connector = JMXConnectorFactory.connect(url); + if (!mailbox.offer(connector)) connector.close(); + } catch (Throwable t) { + mailbox.offer(t); + } + }); + Object result; + try { + result = mailbox.poll(timeout, unit); + if (result == null) { + if (!mailbox.offer("")) result = mailbox.take(); + } + } catch (InterruptedException e) { + throw initCause(new InterruptedIOException(e.getMessage()), e); + } finally { + executor.shutdown(); + } + if (result == null) throw new SocketTimeoutException("Connect timed out: " + url); + if (result instanceof JMXConnector) return (JMXConnector) result; + try { + throw (Throwable) result; + } catch (IOException | RuntimeException | Error e) { + throw e; + } catch (Throwable e) { + // In principle this can't happen but we wrap it anyway + throw new IOException(e.toString(), e); + } + } + + private static T initCause(T wrapper, Throwable wrapped) { + wrapper.initCause(wrapped); + return wrapper; + } + + private static class DaemonThreadFactory implements ThreadFactory { + + public Thread newThread(Runnable r) { + Thread t = Executors.defaultThreadFactory().newThread(r); + t.setDaemon(true); + return t; + } + } + + private static final ThreadFactory daemonThreadFactory = new DaemonThreadFactory(); + + public CanalServerMXBean getCanalServerMXBean() { + if (jmxc == null) { + connect(); + } + return canalServerMXBean; + } + + public void close() { + try { + if (jmxc != null) { + jmxc.close(); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + jmxc = null; + } + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java index 489bdc1b..890d51b9 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java @@ -1,9 +1,29 @@ package com.alibaba.otter.canal.admin.model; +import io.ebean.Finder; + import java.util.Date; -public class CanalConfig { +import javax.persistence.Entity; +import javax.persistence.Id; +@Entity +public class CanalConfig extends Model { + + public static final CanalConfigFinder find = new CanalConfigFinder(); + + public static class CanalConfigFinder extends Finder { + + /** + * Construct using the default EbeanServer. + */ + public CanalConfigFinder(){ + super(CanalConfig.class); + } + + } + + @Id private Long id; private String name; private String content; diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/Model.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/Model.java new file mode 100644 index 00000000..bd0ccfe6 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/Model.java @@ -0,0 +1,35 @@ +package com.alibaba.otter.canal.admin.model; + +import io.ebean.Ebean; +import io.ebean.EbeanServer; +import org.apache.commons.beanutils.PropertyUtils; + +import javax.persistence.MappedSuperclass; +import javax.persistence.OptimisticLockException; + +@MappedSuperclass +public abstract class Model extends io.ebean.Model { + + public void update(String... propertiesNames) { + try { + EbeanServer ebeanServer = Ebean.getDefaultServer(); + Object id = ebeanServer.getBeanId(this); + Object model = ebeanServer.createQuery(this.getClass()).where().idEq(id).findOne(); + for (String propertyName : propertiesNames) { + if (propertyName.startsWith("nn:")) { // not null + propertyName = propertyName.substring(3); + Object val = PropertyUtils.getProperty(this, propertyName); + if (val != null) { + PropertyUtils.setProperty(model, propertyName, val); + } + } else { + Object val = PropertyUtils.getProperty(this, propertyName); + PropertyUtils.setProperty(model, propertyName, val); + } + } + ebeanServer.update(model); + } catch (Exception e) { + throw new OptimisticLockException(e); + } + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java new file mode 100644 index 00000000..c4d4f621 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java @@ -0,0 +1,94 @@ +package com.alibaba.otter.canal.admin.model; + +import java.util.Date; + +import javax.persistence.Entity; +import javax.persistence.Id; + +import io.ebean.Finder; + +@Entity +public class NodeServer extends Model { + + public static final NodeServerFinder find = new NodeServerFinder(); + + public static class NodeServerFinder extends Finder { + + /** + * Construct using the default EbeanServer. + */ + public NodeServerFinder(){ + super(NodeServer.class); + } + + } + + @Id + private Long id; + private String name; + private String ip; + private Integer port; + private Integer port2; + private Integer status = -1; + private Date modifiedTime; + + public void init() { + + } + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getIp() { + return ip; + } + + public void setIp(String ip) { + this.ip = ip; + } + + public Integer getPort() { + return port; + } + + public void setPort(Integer port) { + this.port = port; + } + + public Integer getPort2() { + return port2; + } + + public void setPort2(Integer port2) { + this.port2 = port2; + } + + public Integer getStatus() { + return status; + } + + public void setStatus(Integer status) { + this.status = status; + } + + public Date getModifiedTime() { + return modifiedTime; + } + + public void setModifiedTime(Date modifiedTime) { + this.modifiedTime = modifiedTime; + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/NodeServerService.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/NodeServerService.java new file mode 100644 index 00000000..fd5247ed --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/NodeServerService.java @@ -0,0 +1,20 @@ +package com.alibaba.otter.canal.admin.service; + +import com.alibaba.otter.canal.admin.model.NodeServer; + +import java.util.List; + +public interface NodeServerService { + + void save(NodeServer nodeServer); + + NodeServer detail(Long id); + + void update(NodeServer nodeServer); + + void delete(Long id); + + List findList(NodeServer nodeServer); + + int remoteNodeStatus(String ip, Integer port); +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java index 3e7f2538..9b229b7c 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java @@ -1,28 +1,22 @@ package com.alibaba.otter.canal.admin.service.impl; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; -import com.alibaba.otter.canal.admin.dao.CanalConfigDao; import com.alibaba.otter.canal.admin.model.CanalConfig; import com.alibaba.otter.canal.admin.service.CanalConfigService; -import org.springframework.transaction.annotation.Transactional; @Service public class CanalConfigServiceImpl implements CanalConfigService { - @Autowired - CanalConfigDao canalConfigDao; - public CanalConfig getCanalConfig() { - return canalConfigDao.findById(1L); + return CanalConfig.find.byId(1L); } public CanalConfig getAdapterConfig() { - return canalConfigDao.findById(2L); + return CanalConfig.find.byId(2L); } public void updateContent(CanalConfig canalConfig) { - canalConfigDao.updateContent(canalConfig); + canalConfig.update("content"); } } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java new file mode 100644 index 00000000..305c5f09 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java @@ -0,0 +1,114 @@ +package com.alibaba.otter.canal.admin.service.impl; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.*; + +import com.alibaba.otter.canal.admin.common.exception.ServiceException; +import com.alibaba.otter.canal.admin.jmx.CanalServerMXBean; +import com.alibaba.otter.canal.admin.jmx.JMXConnection; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import com.alibaba.otter.canal.admin.model.NodeServer; +import com.alibaba.otter.canal.admin.service.NodeServerService; + +import io.ebean.Query; + +@Service +public class NodeServerServiceImpl implements NodeServerService { + + private static final Logger logger = LoggerFactory.getLogger(NodeServerServiceImpl.class); + + public void save(NodeServer nodeServer) { + int cnt = NodeServer.find.query() + .where() + .eq("ip", nodeServer.getIp()) + .eq("port", nodeServer.getPort()) + .findCount(); + if (cnt > 0) { + throw new ServiceException("节点信息已存在"); + } + + nodeServer.save(); + + // 检测节点状态 + } + + public NodeServer detail(Long id) { + return NodeServer.find.byId(id); + } + + public void update(NodeServer nodeServer) { + nodeServer.update("name", "ip", "port", "port2"); + + // 检测节点状态 + } + + public void delete(Long id) { + NodeServer nodeServer = NodeServer.find.byId(id); + if (nodeServer != null) { + nodeServer.delete(); + } + } + + public List findList(NodeServer nodeServer) { + Query query = NodeServer.find.query(); + if (nodeServer != null) { + if (StringUtils.isNotEmpty(nodeServer.getName())) { + query.where().like("name", "%" + nodeServer.getName() + "%"); + } + if (StringUtils.isNotEmpty(nodeServer.getIp())) { + query.where().eq("ip", nodeServer.getIp()); + } + } + query.order().asc("id"); + List nodeServers = query.findList(); + + ExecutorService executorService = Executors.newFixedThreadPool(nodeServers.size()); + List> futures = new ArrayList<>(nodeServers.size()); + // 取每个节点的状态 + for (NodeServer ns : nodeServers) { + futures.add(executorService.submit(() -> { + int status = -1; + JMXConnection jmxConnection = new JMXConnection(ns.getIp(), ns.getPort()); + try { + CanalServerMXBean canalServerMXBean = jmxConnection.getCanalServerMXBean(); + status = canalServerMXBean.getStatus(); + } catch (Exception e) { + logger.warn(e.getMessage()); + } finally { + jmxConnection.close(); + } + ns.setStatus(status); + return status != -1; + })); + } + futures.forEach(f -> { + try { + f.get(); + } catch (InterruptedException | ExecutionException e) { + // ignore + } + }); + + executorService.shutdownNow(); + + return nodeServers; + } + + public int remoteNodeStatus(String ip, Integer port) { + JMXConnection jmxConnection = new JMXConnection(ip, port); + try { + CanalServerMXBean canalServerMXBean = jmxConnection.getCanalServerMXBean(); + return canalServerMXBean.getStatus(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + jmxConnection.close(); + } + return -1; + } +} diff --git a/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/BaseTest.java b/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/BaseTest.java new file mode 100644 index 00000000..cc6cc109 --- /dev/null +++ b/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/BaseTest.java @@ -0,0 +1,11 @@ +package com.alibaba.otter.canal.admin; + +import org.junit.runner.RunWith; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.junit4.SpringRunner; + +@RunWith(SpringRunner.class) +@SpringBootTest(classes = { CanalAdminApplication.class }) + +public class BaseTest { +} diff --git a/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/TestApplication.java b/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/TestApplication.java new file mode 100644 index 00000000..2c41f7a2 --- /dev/null +++ b/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/TestApplication.java @@ -0,0 +1,7 @@ +package com.alibaba.otter.canal.admin; + +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class TestApplication { +} diff --git a/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/service/NodeServerServiceTest.java b/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/service/NodeServerServiceTest.java new file mode 100644 index 00000000..d567542b --- /dev/null +++ b/canal-admin/canal-admin-server/src/test/java/com/alibaba/otter/canal/admin/service/NodeServerServiceTest.java @@ -0,0 +1,21 @@ +package com.alibaba.otter.canal.admin.service; + +import com.alibaba.otter.canal.admin.BaseTest; +import com.alibaba.otter.canal.admin.model.NodeServer; +import org.junit.Assert; +import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; + +import java.util.List; + +public class NodeServerServiceTest extends BaseTest { + + @Autowired + NodeServerService nodeServerService; + + @Test + public void findList() { + List list = nodeServerService.findList(null); + Assert.assertNotNull(list); + } +} diff --git a/canal-admin/canal-admin-server/src/test/resources/application.yml b/canal-admin/canal-admin-server/src/test/resources/application.yml new file mode 100644 index 00000000..137af395 --- /dev/null +++ b/canal-admin/canal-admin-server/src/test/resources/application.yml @@ -0,0 +1,12 @@ +spring.datasource: + url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8&useSSL=false + username: root + password: 121212 + driver-class-name: com.mysql.jdbc.Driver + hikari: + maximum-pool-size: 10 + minimum-idle: 1 + +logging: + level: + com.alibaba.otter.canal.admin: DEBUG