Print the steps used in each Cluster Flow pipeline
509904
def clusterflow_pipelines_printout(self):
        
        data = dict()
        html = ''
        for f,d in self.clusterflow_runfiles.items():
            pid = d.get('pipeline_id', 'unknown')
            data[pid] = [
                d.get('pipeline_name'),
                "\n".join(d.get('pipeline_steps', []))
            ]
        for pid, d in data.items():
            html += '''
                <div class="panel panel-default">
                    <div class="panel-heading"><h3 class="panel-title">Pipeline Steps: {} (<code>{}</code>)</h3></div>
                    <pre class="panel-body" style="border:0; background-color:transparent; padding:0 15px; margin:0; color:#666; font-size:90%;">{}</pre>
                </div>
                '''.format(pid, d[0], d[1])
        return html
167556
def process_map
      output    = `ps -eo ppid,pid`
      processes = {}

      output.split("\n").each do |line|
        if result = line.match(/(\d+)\s(\d+)/)
          parent = result[1].to_i
          child  = result[2].to_i

          processes[parent] ||= []
          processes[parent] << child
        end
      end

      processes
    end
1208145
static void showCluster(
      PrintWriter out, boolean advancedView, JobTracker jobTracker) {
    String cluster = "";
    try {
      cluster = JSPUtil.generateClusterResTable(jobTracker);
      if (cluster.equals("")) {
        return;
      }
    } catch (IOException e) {
      return;
    }
    out.print("<h2>Cluster Resource</h2>\n");
    out.print(cluster);
  }
1170209
def list_bootstrap_actions(jobflow_id)
      params = {
        :operation => 'ListBootstrapActions',
        :cluster_id => jobflow_id,
      }
      aws_result = @aws_request.submit(params)
      yield aws_result if block_given?
      JSON.parse(aws_result)
    end
901895
def get_pipeline_names():
  """"""
  class_path_set = set()
  for cls in _PipelineMeta._all_classes:
      if cls.class_path is not None:
        class_path_set.add(cls.class_path)
  return sorted(class_path_set)
201458
def main():
    
    home = os.getcwd()
    options = parse_arguments()
    print(options)
    print()

    #get a list of all directories in the cwd
    dirs = [direc for direc in os.listdir(home) if os.path.isdir(direc)]
    #instantiate all of the classes that we will be using in parallel processing
    instances = []
    for each in dirs:
        thisInstance = meraculousRunAnalyzer(each, options.censor, options.quiet)
        #only process the instances that are meraculous directories
        if thisInstance.isMer:
            instances.append(thisInstance)

    if len(instances) == 0:
        print("There are no meraculous folders in this directory. Exiting")
    elif len(instances) > 0:
        # run the program for each instance
        #process each file sequentially using max number of threads
        #determine the pool size to work with the unique sample names
        pool_size = determine_pool_size(instances)
        pool = ThreadPool(pool_size)
        results = pool.map(run_merReport_dummy, instances)
        pool.close()
        pool.join()
304438
def describe_step(jobflow_id, step_id)
      params = {
        :operation => 'DescribeStep',
        :cluster_id => jobflow_id,
        :step_id => step_id
      }
      aws_result = @aws_request.submit(params)
      yield aws_result if block_given?
      JSON.parse(aws_result)
    end
764003
def list_stacks(awsclient):
    
    client_cf = awsclient.get_client('cloudformation')
    response = client_cf.list_stacks(
        StackStatusFilter=[
            'CREATE_IN_PROGRESS', 'CREATE_COMPLETE', 'ROLLBACK_IN_PROGRESS',
            'ROLLBACK_COMPLETE', 'DELETE_IN_PROGRESS', 'DELETE_FAILED',
            'UPDATE_IN_PROGRESS', 'UPDATE_COMPLETE_CLEANUP_IN_PROGRESS',
            'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_IN_PROGRESS',
            'UPDATE_ROLLBACK_FAILED',
            'UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS',
            'UPDATE_ROLLBACK_COMPLETE',
        ]
    )
    result = {}
    stack_sum = 0
    for summary in response['StackSummaries']:
        result['StackName'] = summary["StackName"]
        result['CreationTime'] = summary['CreationTime']
        result['StackStatus'] = summary['StackStatus']
        print(json2table(result))
        stack_sum += 1
    print('listed %s stacks' % str(stack_sum))
745690
def downstream(self, step_name):
        """"""
        return list(self.steps[dep] for dep in self.dag.downstream(step_name))
1138778
def get_outputs(self):
    
    assert self.SUCCESS == self.get_status()
    ss = model.ShardState.find_all_by_mapreduce_state(self._state)
    for s in ss:
      yield iter(s.writer_state.get("outs", []))
623088
def ReadAllFlowObjects(
      self,
      client_id = None,
      min_create_time = None,
      max_create_time = None,
      include_child_flows = True,
  ):
    """"""
    res = []
    for flow in itervalues(self.flows):
      if ((client_id is None or flow.client_id == client_id) and
          (min_create_time is None or flow.create_time >= min_create_time) and
          (max_create_time is None or flow.create_time <= max_create_time) and
          (include_child_flows or not flow.parent_flow_id)):
        res.append(flow.Copy())
    return res
1105377
function getIndexStats(done, results) {
  var client = results.client;
  var ns = mongodbNS(results.namespace);
  var pipeline = [
    { $indexStats: {} },
    {
      $project: {
        name: 1,
        usageHost: '$host',
        usageCount: '$accesses.ops',
        usageSince: '$accesses.since'
      }
    }
  ];
  var collection = client.db(ns.database).collection(ns.collection);
  collection.aggregate(pipeline, { cursor: {} }).toArray(function(err, res) {
    if (err) {
      if (isNotAuthorizedError(err)) {
        /**
         * In the 3.2 server, `readWriteAnyDatabase@admin` does not grant sufficient privileges for $indexStats.
         * The `clusterMonitor` role is required to run $indexStats.
         * @see https://jira.mongodb.org/browse/INT-1520
         */
        return done(null, {});
      }

      if (err.message.match(/Unrecognized pipeline stage name/)) {
        // $indexStats not yet supported, return empty document
        return done(null, {});
      }
      done(err);
    }
    res = _.mapKeys(res, function(stat) {
      return stat.name;
    });
    done(null, res);
  });
}
1364832
def run_steps(node: Node, pipeline: RenderingPipeline, **args):
    """"""
    for step in pipeline.steps:
        result = step(node, pipeline=pipeline, **args)
        if isinstance(result, dict):
            args = {**args, **result}
1087603
def generate_pipeline_code(pipeline_tree, operators):
    
    steps = _process_operator(pipeline_tree, operators)
    pipeline_text = "make_pipeline(\n{STEPS}\n)".format(STEPS=_indent(",\n".join(steps), 4))
    return pipeline_text
1145376
def get_outputs(self):
        
        end_p = [p for p in self.postorder() if not self.incoming_edges(p)]
        self.log.debug('%s got output pipers %s' % (repr(self), end_p))
        return end_p
1291617
def process(self, document):
        """"""
        self.logger.info("Processing group of tasks (parallel=%s)", self.get_parallel_mode())
        self.pipeline.data.env_list[2] = {}

        output, shells = [], []
        result = Adapter({'success': True, 'output': []})
        for task_entry in document:
            key, entry = list(task_entry.items())[0]

            if (not self.parallel or key == 'env') and len(shells) > 0:
                result = Adapter(self.process_shells(shells))
                output += result.output
                shells = []
                if not result.success:
                    break

            if key == 'env':
                self.pipeline.data.env_list[2].update(entry)

            elif key in ['shell', 'docker(container)', 'docker(image)', 'python',
                         'packer', 'ansible(simple)']:
                self.prepare_shell_data(shells, key, entry)

        if result.success:
            result = Adapter(self.process_shells(shells))
            output += result.output

        self.event.delegate(result.success)
        return {'success': result.success, 'output': output}
341691
def _walk_through(job_dir, display_progress=False):
    '''
    
    '''
    serial = salt.payload.Serial(__opts__)

    for top in os.listdir(job_dir):
        t_path = os.path.join(job_dir, top)

        for final in os.listdir(t_path):
            load_path = os.path.join(t_path, final, '.load.p')
            with salt.utils.files.fopen(load_path, 'rb') as rfh:
                job = serial.load(rfh)

            if not os.path.isfile(load_path):
                continue

            with salt.utils.files.fopen(load_path, 'rb') as rfh:
                job = serial.load(rfh)
            jid = job['jid']
            if display_progress:
                __jid_event__.fire_event(
                    {'message': 'Found JID {0}'.format(jid)},
                    'progress'
                )
            yield jid, job, t_path, final