根据pipelineId找到pipeline
547327
public Pipeline findById(Long pipelineId) {

        Assert.assertNotNull(pipelineId);
        List<Pipeline> pipeline = listByIds(pipelineId);

        if (pipeline.size() != 1) {
            String exceptionCause = "query pipeline by pipelineId:" + pipelineId + " but return " + pipeline.size()
                                    + " pipeline!";
            logger.error("ERROR ## " + exceptionCause);
            throw new ManagerException(exceptionCause);
        }
        return pipeline.get(0);
    }
1273740
def pipeline_get(id, hosts=None, profile=None):
    '''
    
    '''
    es = _get_instance(hosts, profile)

    try:
        return es.ingest.get_pipeline(id=id)
    except elasticsearch.NotFoundError:
        return None
    except elasticsearch.TransportError as e:
        raise CommandExecutionError("Cannot create pipeline {0}, server returned code {1} with message {2}".format(id, e.status_code, e.error))
    except AttributeError:
        raise CommandExecutionError("Method is applicable only for Elasticsearch 5.0+")
652930
public Map<Long, DelayStatInfo> listTimelineDelayStat(Long pipelineId, Date start, Date end) {

        Map<Long, DelayStatInfo> delayStatInfos = new LinkedHashMap<Long, DelayStatInfo>();
        List<DelayStatDO> delayStatDOs = delayStatDao.listTimelineDelayStatsByPipelineId(pipelineId, start, end);
        int size = delayStatDOs.size();
        int k = size - 1;
        for (Long i = start.getTime(); i <= end.getTime(); i += 60 * 1000) {
            DelayStatInfo delayStatInfo = new DelayStatInfo();
            List<DelayStat> delayStats = new ArrayList<DelayStat>();
            // 取出每个时间点i以内的数据，k是一个游标，每次遍历时前面已经取过了的数据就不用再遍历了
            for (int j = k; j >= 0; --j) {
                if ((i - delayStatDOs.get(j).getGmtModified().getTime() <= 60 * 1000)
                    && (i - delayStatDOs.get(j).getGmtModified().getTime() >= 0)) {
                    delayStats.add(delayStatDOToModel(delayStatDOs.get(j)));
                    k = j - 1;
                }// 如果不满足if条件，则后面的数据也不用再遍历
                else {
                    break;
                }
            }
            if (delayStats.size() > 0) {
                delayStatInfo.setItems(delayStats);
                delayStatInfos.put(i, delayStatInfo);
            }

        }
        return delayStatInfos;
    }
234460
def activate_pipeline(self):
        """"""
        self.client.activate_pipeline(pipelineId=self.pipeline_id)
        LOG.info("Activated Pipeline %s", self.pipeline_id)
431583
function countChildren(pipelineId) {
  var current = STATUS_MAP.pipelines[pipelineId];
  if (!current) {
    return [0, 0];
  }
  var total = 1;
  var done = 0;
  if (current.status == 'done') {
    done += 1;
  }
  for (var i = 0, n = current.children.length; i < n; i++) {
    var parts = countChildren(current.children[i]);
    total += parts[0];
    done += parts[1];
  }
  return [total, done];
}
1044140
public List<Channel> listByNodeId(Long nodeId, ChannelStatus... statuses) {
        List<Channel> channels = new ArrayList<Channel>();
        List<Channel> results = new ArrayList<Channel>();
        try {
            List<Pipeline> pipelines = pipelineService.listByNodeId(nodeId);
            List<Long> pipelineIds = new ArrayList<Long>();
            for (Pipeline pipeline : pipelines) {
                pipelineIds.add(pipeline.getId());
            }

            if (pipelineIds.isEmpty()) { // 没有关联任务直接返回
                return channels;
            }

            // 反查对应的channel
            channels = listByPipelineIds(pipelineIds.toArray(new Long[pipelineIds.size()]));
            if (null == statuses || statuses.length == 0) {
                return channels;
            }

            for (Channel channel : channels) {
                for (ChannelStatus status : statuses) {
                    if (channel.getStatus().equals(status)) {
                        results.add(channel);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("ERROR ## list query channel by nodeId:" + nodeId + " has an exception!");
            throw new ManagerException(e);
        }
        return results;
    }
1338527
public static String getProcess(Long channelId, Long pipelineId, Long processId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_PROCESS_FORMAT,
            String.valueOf(channelId),
            String.valueOf(pipelineId),
            getProcessNode(processId));
    }
972294
public DelayStat findRealtimeDelayStat(Long pipelineId) {
        Assert.assertNotNull(pipelineId);
        DelayStatDO delayStatDO = delayStatDao.findRealtimeDelayStat(pipelineId);
        DelayStat delayStat = new DelayStat();
        if (delayStatDO != null) {
            delayStat = delayStatDOToModel(delayStatDO);
        }
        return delayStat;
    }
447607
public void init(Long channelId, Long pipelineId) {
        String path = ManagePathUtils.getPipeline(channelId, pipelineId);
        String processRootPath = ManagePathUtils.getProcessRoot(channelId, pipelineId);
        String terminRootPath = ManagePathUtils.getTerminRoot(channelId, pipelineId);
        String remedyRootPath = ManagePathUtils.getRemedyRoot(channelId, pipelineId);
        String lockRootPath = ManagePathUtils.getLockRoot(channelId, pipelineId);
        String loadLockPath = lockRootPath + "/" + ArbitrateConstants.NODE_LOCK_LOAD;
        try {
            zookeeper.createPersistent(path, true);//创建父节点
            zookeeper.create(processRootPath, new byte[0], CreateMode.PERSISTENT);
            zookeeper.create(terminRootPath, new byte[0], CreateMode.PERSISTENT);
            zookeeper.create(remedyRootPath, new byte[0], CreateMode.PERSISTENT);
            zookeeper.create(lockRootPath, new byte[0], CreateMode.PERSISTENT);
            zookeeper.create(loadLockPath, new byte[0], CreateMode.PERSISTENT);
        } catch (ZkNodeExistsException e) {
            // 如果节点已经存在，则不抛异常
            // ignore
        } catch (ZkException e) {
            throw new ArbitrateException("Pipeline_init", pipelineId.toString(), e);
        }
    }
646670
public List<Pipeline> listByChannelIds(Long... channelId) {
        Assert.assertNotNull(channelId);
        List<Pipeline> pipelines = new ArrayList<Pipeline>();
        try {

            List<PipelineDO> pipelineDos = pipelineDao.listByChannelIds(channelId);
            if (pipelineDos.isEmpty()) {
                logger.debug("DEBUG ## query pipeline by channelId:" + channelId + " return null.");
                return pipelines;
            }
            pipelines = doToModel(pipelineDos);
        } catch (Exception e) {
            logger.error("ERROR ## query pipelines by channelIds:" + channelId.toString() + " has an exception!");
            throw new ManagerException(e);
        }
        return pipelines;
    }
288706
public void waitForPermit(Long pipelineId) throws InterruptedException {
        Assert.notNull(pipelineId);

        PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
        permitMonitor.waitForPermit();// 阻塞等待授权
    }
502414
public static String getLockRoot(Long channelId, Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_LOCK_ROOT,
            String.valueOf(channelId),
            String.valueOf(pipelineId));
    }
1307713
def get_pipeline_id(self):
        """"""

        all_pipelines = []
        paginiator = self.client.get_paginator('list_pipelines')
        for page in paginiator.paginate():
            all_pipelines.extend(page['pipelineIdList'])

        for pipeline in all_pipelines:
            if pipeline['name'] == self.datapipeline_data.get('name', self.app_name):
                self.pipeline_id = pipeline['id']
                LOG.info("Pipeline ID Found")
                return
        LOG.info("Pipeline ID Not Found for %s", self.app_name)
1135060
function pipe(file, type, keep) {
  var processors = [];
  var prop = file[type];

  if (type === 'standard' && 'undefined' === typeof prop) {
    processors.push('builtin');
  }

  if (prop) {
    var typeOf = typeof prop;
    if (typeOf === 'string') {
      prop = prop.trim().split(/\s*,\s*/);
    } else if (!Array.isArray(prop)) {
      prop = [prop];
    }

    processors = processors.concat(prop);
  }

  fis.emit('compile:' + type, file);

  if (processors.length) {

    // 过滤掉同名的插件, 没必要重复操作。
    processors = processors.filter(function(item, idx, arr) {
      item = item.__name || item;

      return idx === _.findIndex(arr, function(target) {
        target = target.__name || target;

        return target === item;
      });
    });

    var callback = function(processor, settings, key) {
      settings.filename = file.realpath;
      var content = file.getContent();
      try {
        fis.log.debug('pipe [' + key + '] start');
        var result = processor(content, file, settings);
        fis.log.debug('pipe [' + key + '] end');
        if (keep) {
          file.setContent(content);
        } else if (typeof result === 'undefined') {
          fis.log.warning('invalid content return of pipe [' + key + ']');
        } else {
          file.setContent(result);
        }
      } catch (e) {
        if (typeof e === 'string') {
          e = new Error(e);
        }

        //log error
        fis.log.debug('pipe [' + key + '] fail');
        var msg = key + ': ' + String(e.message || e.msg || e).trim() + ' [' + (e.filename || file.realpath);
        if (e.hasOwnProperty('line')) {
          msg += ':' + e.line;
          if (e.hasOwnProperty('col')) {
            msg += ':' + e.col;
          } else if (e.hasOwnProperty('column')) {
            msg += ':' + e.column;
          }
        }
        msg += ']';
        e.message = msg;
        error(e);
      }
    };

    processors.forEach(function(processor, index) {
      var typeOf = typeof processor,
        key, options;

      if (typeOf === 'object' && processor.__name) {
        options = processor;
        processor = processor.__name;
        typeOf = typeof processor;
      }

      if (typeOf === 'string') {
        key = type + '.' + processor;
        processor = (type === 'standard' && processor === 'builtin') ? builtinStandard : fis.require(type, processor);
      } else {
        key = type + '.' + index;
      }

      if (typeof processor === 'function') {
        var settings = {};
        _.assign(settings, processor.defaultOptions || processor.options || {});
        _.assign(settings, fis.media().get('settings.' + key, {}));
        _.assign(settings, options || {});

        // 删除隐藏配置
        delete settings.__name;
        delete settings.__plugin;
        delete settings.__pos;
        delete settings.__isPlugin;

        callback(processor, settings, key);
      } else {
        fis.log.warning('invalid processor [modules.' + key + ']');
      }
    });
  }
}
607163
def post_pipeline(self, pipeline):
        
        if isinstance(pipeline, str):
            pipeline_str = pipeline
        else:
            pipeline_str = json.dumps(pipeline)

        pipeline_json = json.loads(pipeline_str)

        # Note pipeline name is manual
        name = '{0} (onetime-{1})'.format(pipeline_json['name'], self.environments[0])
        pipeline_json['name'] = name

        # Inject pipeline Id so that it does not override existing pipelines
        pipeline_id = super().compare_with_existing(onetime=True)
        if pipeline_id:
            pipeline_json['id'] = pipeline_id
        else:
            del pipeline_json['id']

        # disable trigger as not to accidently kick off multiple deployments
        for trigger in pipeline_json['triggers']:
            trigger['enabled'] = False

        self.log.debug('Manual Pipeline JSON:\n%s', pipeline_json)
        super().post_pipeline(pipeline_json)
492830
function isRunByHusky (ppid) {
  return find('pid', ppid).then(list => {
    debug('process list: %o', list)

    const cmd = list[0] && list[0].cmd
    const pppid = list[0] && list[0].ppid

    if (!cmd) {
      return false
    }

    if (huskyCmdReg.test(cmd)) {
      return true
    }

    // 一直迭代到最顶
    return isRunByHusky(pppid)
  })
}
170287
def delete_all_pipelines(self):
		'''
		'''
		code, data = self.get_pipeline()
		if code == requests.codes.ok:
			for pl_data in data:
				c, d = self.delete_pipeline(pl_data['pipelineKey'])
				if c != requests.codes.ok:
					code = c
					data = d
		return code, data