content: /*
   * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.alibaba.otter.node.etl;
  
  import java.lang.management.ManagementFactory;
  import java.lang.management.MemoryUsage;
  import java.lang.management.OperatingSystemMXBean;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.ThreadPoolExecutor;
  
  import org.apache.commons.lang.StringUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.node.common.config.ConfigClientService;
  import com.alibaba.otter.node.common.config.NodeTaskListener;
  import com.alibaba.otter.node.common.config.NodeTaskService;
  import com.alibaba.otter.node.common.config.model.NodeTask;
  import com.alibaba.otter.node.common.config.model.NodeTask.TaskEvent;
  import com.alibaba.otter.node.etl.common.datasource.DataSourceService;
  import com.alibaba.otter.node.etl.common.db.dialect.DbDialectFactory;
  import com.alibaba.otter.node.etl.common.jmx.StageAggregationCollector;
  import com.alibaba.otter.node.etl.common.task.GlobalTask;
  import com.alibaba.otter.node.etl.extract.ExtractTask;
  import com.alibaba.otter.node.etl.load.LoadTask;
  import com.alibaba.otter.node.etl.select.SelectTask;
  import com.alibaba.otter.node.etl.transform.TransformTask;
  import com.alibaba.otter.shared.arbitrate.ArbitrateEventService;
  import com.alibaba.otter.shared.arbitrate.ArbitrateManageService;
  import com.alibaba.otter.shared.arbitrate.impl.manage.NodeSessionExpired;
  import com.alibaba.otter.shared.arbitrate.impl.zookeeper.ZooKeeperClient;
  import com.alibaba.otter.shared.common.model.config.ConfigException;
  import com.alibaba.otter.shared.common.model.config.enums.StageType;
  import com.alibaba.otter.shared.common.model.config.node.Node;
  import com.alibaba.otter.shared.common.model.config.pipeline.Pipeline;
  import com.alibaba.otter.shared.common.utils.AddressUtils;
  import com.alibaba.otter.shared.common.utils.JsonUtils;
  import com.alibaba.otter.shared.common.utils.version.VersionInfo;
  import com.google.common.base.Function;
  import com.google.common.collect.MapMaker;
  import com.google.common.collect.OtterMigrateMap;
  
  /**
   * 管理和维护对应node机器内的S.E.T.L任务，实时接收manager推送的NodeTask调度信息，可查看 {@linkplain NodeTaskService}
   * 
   * @author jianghang 2012-4-21 下午04:48:12
   * @version 4.0.2
   */
  public class OtterController implements NodeTaskListener, OtterControllerMBean {
  
      private static final Logger                   logger      = LoggerFactory.getLogger(OtterController.class);
  
      // 第一层为pipelineId，第二层为S.E.T.L模块
      private Map<Long, Map<StageType, GlobalTask>> controllers = OtterMigrateMap.makeComputingMap(new Function<Long, Map<StageType, GlobalTask>>() {
  
                                                                    public Map<StageType, GlobalTask> apply(Long pipelineId) {
                                                                        return new MapMaker().makeMap();
                                                                    }
                                                                });
      private ConfigClientService                   configClientService;
      private ArbitrateManageService                arbitrateManageService;
      private NodeTaskService                       nodeTaskService;
      // 各种资源管理
      private DataSourceService                     dataSourceService;                                           // 连接池资源
      private DbDialectFactory                      dbDialectFactory;                                            // 数据库信息资源
      private ArbitrateEventService                 arbitrateEventService;                                       // 仲裁器资源
      private ExecutorService                       executorService;
  
      private StageAggregationCollector             stageAggregationCollector;
  
      public void start() throws Throwable {
          // 初始化节点
          initNid();
          nodeTaskService.addListener(this); // 将自己添加为NodeTask响应者
      }
  
      public void stop() throws Throwable {
          for (Map<StageType, GlobalTask> tasks : controllers.values()) {
              for (GlobalTask task : tasks.values()) {
                  try {
                      task.shutdown();
                  } catch (Exception e) {
                      logger.error("##shutdown task error!", e);
                  }
              }
          }
  
          try {
              Long nid = configClientService.currentNode().getId();
              arbitrateManageService.nodeEvent().destory(Long.valueOf(nid));
          } catch (Exception e) {
              logger.error("##destory node error!", e);
          }
  
          try {
              arbitrateEventService.toolEvent().release();
          } catch (Exception e) {
              logger.error("##destory arbitrate error!", e);
          }
  
          try {
              nodeTaskService.stopNode(); // 通知manager停止当前node
          } catch (Exception e) {
              logger.error("##stop node error!", e);
          }
  
          try {
              OtterContextLocator.close();
          } catch (Exception e) {
              logger.error("##cloes spring error!", e);
          }
  
          ZooKeeperClient.destory();// 关闭zookeeper
      }
  
      public boolean process(List<NodeTask> nodeTasks) {
          if (nodeTasks == null || nodeTasks.isEmpty()) {
              return true;
          }
  
          for (NodeTask nodeTask : nodeTasks) {
              boolean shutdown = nodeTask.isShutdown();
              Long pipelineId = nodeTask.getPipeline().getId();
              if (shutdown) {
                  Map<StageType, GlobalTask> tasks = controllers.remove(pipelineId);
                  if (tasks != null) {
                      logger.info("INFO ## shutdown this pipeline sync ,the pipelineId = {} and tasks = {}", pipelineId,
                                  tasks.keySet());
                      stopPipeline(pipelineId, tasks);
                  } else {
                      logger.info("INFO ## this pipeline id = {} is not start sync", pipelineId);
                  }
              } else {
                  startPipeline(nodeTask);
              }
          }
  
          return true;
      }
  
      // ===================== helper method ======================
  
      public void startPipeline(NodeTask nodeTask) {
          Long pipelineId = nodeTask.getPipeline().getId();
          releasePipeline(pipelineId);
          Map<StageType, GlobalTask> tasks = controllers.get(pipelineId);
          // 处理具体的任务命令
          List<StageType> stage = nodeTask.getStage();
          List<TaskEvent> event = nodeTask.getEvent();
          for (int i = 0; i < stage.size(); i++) {
              StageType stageType = stage.get(i);
              TaskEvent taskEvent = event.get(i);
              if (taskEvent.isCreate()) {
                  startTask(nodeTask.getPipeline(), tasks, stageType);
              } else {
                  stopTask(tasks, stageType);
              }
          }
      }
  
      private void startTask(Pipeline pipeline, Map<StageType, GlobalTask> tasks, StageType taskType) {
          if (tasks.get(taskType) != null && tasks.get(taskType).isAlive()) {
              logger.warn("WARN ## this task = {} has started", taskType);
          }
  
          GlobalTask task = null;
          if (taskType.isSelect()) {
              task = new SelectTask(pipeline.getId());
          } else if (taskType.isExtract()) {
              task = new ExtractTask(pipeline.getId());
          } else if (taskType.isTransform()) {
              task = new TransformTask(pipeline.getId());
          } else if (taskType.isLoad()) {
              task = new LoadTask(pipeline.getId());
          }
  
          if (task != null) {
              OtterContextLocator.autowire(task); // 注入一下spring资源
              task.start();
              tasks.put(taskType, task);
              logger.info("INFO ## start this task = {} success", taskType.toString());
          }
      }
  
      private void stopTask(Map<StageType, GlobalTask> tasks, StageType taskType) {
          GlobalTask task = tasks.remove(taskType);
          if (task != null) {
              task.shutdown();
              logger.info("INFO ## taskName = {} has shutdown", taskType);
          } else {
              logger.info("INFo ## taskName = {} is not started", taskType);
          }
  
      }
  
      private void stopPipeline(Long pipelineId, Map<StageType, GlobalTask> tasks) {
          for (GlobalTask task : tasks.values()) {
              try {
                  task.shutdown();
              } catch (Exception e) {
                  logger.error("## stop s/e/t/l task error!", e);
              } finally {
                  tasks.remove(task);
              }
          }
          // close other resources.
          try {
              Thread.sleep(1 * 1000); // sleep 5s，等待S.E.T.L释放线程
          } catch (InterruptedException e) {
              logger.error("ERROR ## ", e);
          }
  
          // 释放资源
          releasePipeline(pipelineId);
          arbitrateEventService.toolEvent().release(pipelineId);
      }
  
      private void releasePipeline(Long pipelineId) {
          dataSourceService.destroy(pipelineId);
          dbDialectFactory.destory(pipelineId);
      }
  
      private void initNid() {
          // 获取一下nid变量
          String nid = System.getProperty(OtterConstants.NID_NAME);
          if (StringUtils.isEmpty(nid)) {
              throw new ConfigException("nid is not set!");
          }
          logger.info("INFO ## the nodeId = {}", nid);
          checkNidVaild(nid);
          arbitrateManageService.nodeEvent().init(Long.valueOf(nid));
          // 添加session expired处理
          NodeSessionExpired sessionExpired = new NodeSessionExpired();
          sessionExpired.setNodeEvent(arbitrateManageService.nodeEvent());
          ZooKeeperClient.registerNotification(sessionExpired);
      }
  
      // 判断本机ip是否和node.getIp()相同
      private void checkNidVaild(String nid) {
          Node node = configClientService.currentNode();
          String hostIp = AddressUtils.getHostIp();
          String nodeIp = node.getIp();
          int nodePort = node.getPort().intValue();
          if (!AddressUtils.isHostIp(nodeIp)) {
              throw new IllegalArgumentException(
                                                 String.format("node[%s] ip[%s] port[%s] , but your host ip[%s] is not matched!",
                                                               nid, nodeIp, nodePort, hostIp));
          }
      }
  
      // ================ mbean info =======================
  
      public String getHeapMemoryUsage() {
          MemoryUsage memoryUsage = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage();
          return JsonUtils.marshalToString(memoryUsage);
      }
  
      public String getNodeSystemInfo() {
          OperatingSystemMXBean mbean = ManagementFactory.getOperatingSystemMXBean();
          StringBuilder buf = new StringBuilder();
          buf.append("").append(mbean.getName()).append(' ').append(mbean.getVersion()).append(' ').append(mbean.getArch());
          buf.append(" @ ").append(mbean.getAvailableProcessors()).append(" cores");
          buf.append(" , 【 load average:").append(mbean.getSystemLoadAverage()).append(" 】");
          return buf.toString();
      }
  
      public String getNodeVersionInfo() {
          return VersionInfo.getVersion() + " [ r" + VersionInfo.getRevision() + " ] @ " + VersionInfo.getDate();
      }
  
      public int getRunningPipelineCount() {
          return controllers.size();
      }
  
      public List<Long> getRunningPipelines() {
          return new ArrayList<Long>(controllers.keySet());
      }
  
      public int getThreadActiveSize() {
          if (executorService instanceof ThreadPoolExecutor) {
              ThreadPoolExecutor pool = (ThreadPoolExecutor) executorService;
              return pool.getActiveCount();
          }
  
          return 0;
      }
  
      public int getThreadPoolSize() {
          if (executorService instanceof ThreadPoolExecutor) {
              ThreadPoolExecutor pool = (ThreadPoolExecutor) executorService;
              return pool.getCorePoolSize();
          }
  
          return 0;
      }
  
      public void setThreadPoolSize(int size) {
          if (executorService instanceof ThreadPoolExecutor) {
              ThreadPoolExecutor pool = (ThreadPoolExecutor) executorService;
              pool.setCorePoolSize(size);
              pool.setMaximumPoolSize(size);
          }
      }
  
      public void setProfile(boolean profile) {
          stageAggregationCollector.setProfiling(profile);
      }
  
      public boolean isSelectRunning(Long pipelineId) {
          return controllers.get(pipelineId).containsKey(StageType.SELECT);
      }
  
      public boolean isExtractRunning(Long pipelineId) {
          return controllers.get(pipelineId).containsKey(StageType.EXTRACT);
      }
  
      public boolean isTransformRunning(Long pipelineId) {
          return controllers.get(pipelineId).containsKey(StageType.TRANSFORM);
      }
  
      public boolean isLoadRunning(Long pipelineId) {
          return controllers.get(pipelineId).containsKey(StageType.LOAD);
      }
  
      public String selectStageAggregation(Long pipelineId) {
          return stageAggregationCollector.histogram(pipelineId, StageType.SELECT);
      }
  
      public String extractStageAggregation(Long pipelineId) {
          return stageAggregationCollector.histogram(pipelineId, StageType.EXTRACT);
      }
  
      public String transformStageAggregation(Long pipelineId) {
          return stageAggregationCollector.histogram(pipelineId, StageType.TRANSFORM);
      }
  
      public String loadStageAggregation(Long pipelineId) {
          return stageAggregationCollector.histogram(pipelineId, StageType.LOAD);
      }
  
      public String selectPendingProcess(Long pipelineId) {
          return pendingProcess(pipelineId, StageType.SELECT);
      }
  
      public String extractPendingProcess(Long pipelineId) {
          return pendingProcess(pipelineId, StageType.EXTRACT);
      }
  
      public String transformPendingProcess(Long pipelineId) {
          return pendingProcess(pipelineId, StageType.TRANSFORM);
      }
  
      public String loadPendingProcess(Long pipelineId) {
          return pendingProcess(pipelineId, StageType.LOAD);
      }
  
      private String pendingProcess(Long pipelineId, StageType stage) {
          GlobalTask task = controllers.get(pipelineId).get(stage);
          if (task != null) {
              return "stage:" + stage + " , pending:[" + StringUtils.join(task.getPendingProcess(), ',') + "]";
          } else {
              return "node don't running stage:" + stage;
          }
      }
  
      // ==================== setter / getter =======================
  
      public void setNodeTaskService(NodeTaskService nodeTaskService) {
          this.nodeTaskService = nodeTaskService;
      }
  
      public void setConfigClientService(ConfigClientService configClientService) {
          this.configClientService = configClientService;
      }
  
      public void setArbitrateManageService(ArbitrateManageService arbitrateManageService) {
          this.arbitrateManageService = arbitrateManageService;
      }
  
      public void setDataSourceService(DataSourceService dataSourceService) {
          this.dataSourceService = dataSourceService;
      }
  
      public void setDbDialectFactory(DbDialectFactory dbDialectFactory) {
          this.dbDialectFactory = dbDialectFactory;
      }
  
      public void setArbitrateEventService(ArbitrateEventService arbitrateEventService) {
          this.arbitrateEventService = arbitrateEventService;
      }
  
      public void setStageAggregationCollector(StageAggregationCollector stageAggregationCollector) {
          this.stageAggregationCollector = stageAggregationCollector;
      }
  
      public void setExecutorService(ExecutorService executorService) {
          this.executorService = executorService;
      }
  
  }
  