Class: Zillabyte::Command::Flows

# File 'lib/zillabyte/cli/flows.rb', line 168

def authorize

  id = options[:id] || shift_argument
  if id.nil?
    id = read_name_from_conf(options)
  end
  make_public = options[:public]
  make_private = options[:private]
  
  error("no id given", type) if id.nil?
  error("both --public and --private cannot be given", type) if make_public && make_private
  scope = "public"
  if make_private
    scope = "private"
  end

  res = self.api.request(
    :expects  => 200,
    :method   => :post,
    :path     => "/flows/#{CGI.escape(id)}/authorizations",
    :body     => {:scope => scope}.to_json
  ).body

  display "Authorization updated"
end

#delete ⇒ `Object`

flows:delete ID

Deletes a flow. If the flow is running, this command will kill it.

-f, –force # Don’t ask for confirmation –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 617

def delete
  id = options[:id] || shift_argument

  if id.nil?
    id = read_name_from_conf(options)
  end
  forced = options[:force]
  type = options[:output_type]

  if not forced

    if !type.nil?
      error("specify -f, --force to confirm deletion", type)
    end

    while true
      display "This operation cannot be undone. Are you sure you want to delete this flow? (yes/no):", false
      confirm = ask
      break if confirm == "yes" || confirm == "no"
      display "Please enter 'yes' to delete the flow or 'no' to exit"
    end

  end

  confirmed = forced || confirm == "yes"

  if confirmed
    display "Destroying flow ##{id}...please wait..." if type.nil?
    response = api.flows.create_delete(id, options)

    if response["job_id"]
      options[:job_id] = response["job_id"]
      flow_id = response["flow_id"]

      start = Time.now.utc
      display "Destroy request sent. Please wait..." if type.nil?

      while(Time.now.utc < start + MAX_POLL_SECONDS) do

      # Poll
        res = self.api.flows.poll_delete(flow_id, options)

        case res['status']
        when 'completed'
          if res['return']
            if type == "json"
              display "{}"
            else
              display "Flow ##{id} destroyed"
            end
          else
            throw "something is wrong: #{res}"
          end
          # success! continue below
          break
        when 'running'
          sleep(POLL_SLEEP)
        else
          throw "unknown status: #{res}"
        end

      end
    elsif response["error"]
      error(response["error"], type)
    else
      error("remote server error (f413)", type)
    end

  end

end

#details ⇒ `Object`

flows:details ID

List details for a flow.

–output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 357

def details

  flow_id = options[:id] || shift_argument
  type = options[:output_type]

  if flow_id.nil?
    flow_id = read_name_from_conf(options)
  end

  res = api.request(
    :expects => 200,
    :method => :get,
    :path => "/flows/#{flow_id}/details",
    :body => options.to_json
  )

  return if res.body.nil?
  error(res.body["error"], type) if res.body["error"]
  return if res.body["state"].nil?

  flow_state = res.body["state"]
  if type.nil?
    display "State: #{flow_state}"
  elsif ["KILLED", "RETIRED"].member? flow_state
    display res.body.to_json
    return
  end

  return if res.body["instances"].nil?
  instances = res.body["instances"].select do |instance|
    !instance.nil?
  end
  if instances.empty?
    if type
      display res.body.to_json
    end
    return
  end

  # a nested hash so can't map directly
  # instead i'll map separately and then merge the arrays
  heading1 = ["type", "name"]

  # a level deeper
  heading2 = ["consumed", "emitted", "errors", "parallelism"]

  # for the table output
  headings = heading1 + heading2

  # get the values for the headings
  type_name = instances.map do |instance|

    heading1.map do |heading|
      instance[heading]
    end

  end

  # get the values for the headings
  stats = instances.map do |instance|


    heading2.map do |heading|
      instance["stats"][heading]
    end



  end

  # combine the arrays
  rows = type_name.zip(stats).map{|x,y| x.concat y}

  # check if it's from a dataset.
  # if the source is a dataset, there's a sharder. 
  # remove source types, and hardcode sharder-named instances as source

  from_dataset = false

  rows.each do | row |
    if row[1].scan("sharder").length != 0
      from_dataset = true
      break
    end
  end

  if from_dataset == true

    rows = rows.each do |row|
      if row[0].scan("source").length != 0
        rows.delete(row)
      end

    end


    rows = rows.each do |row|
      if row[1].scan("sharder").length != 0
        row[0] = "source"
      end
    end

  end

  # rows to be output to the CLI
  rows = rows.group_by { |type, name, consumed, emitted, errors, parallelism| name }.map{ |hashkey, array| [array[0][0], hashkey, array.inject(0){ |sum, i | sum + i[2].to_i }, array.inject(0){ |sum, i | sum + i[3].to_i },array.inject(0){ |sum, i | sum + i[4].to_i }, array.inject(0){|value, i| i[5].to_i}]}
  # example sums: [["sink", "has_facebook_likebox_and_vwo", 73, 0, 0], ["source", "sharder.1", 16, 10367872, 1], ["each", "each_3", 10367872, 73, 0]]

  # calculate percentage for the user to see

  attempts = 0
  total = 0

  if from_dataset == true

    rows.each do |instance|
      if instance[0] == "source"
        total = instance[3]
      end
    end


    rows.each do |instance|
      if instance[0] == "each"
        attempts = instance[2]
      end
    end
  end

  if attempts > 0 && total > 0
    completion = attempts/total*100
  else
    completion = 0
  end

  # output on the CLI
  if type.nil?
    require('colorize')
    require("zillabyte/cli/helpers/table_output_builder")
    display TableOutputBuilder.build_terminal_table(headings, rows)
    if from_dataset == true
      display "percent complete: #{completion}%\n"
    end
    display "To view error details, type "+"`zillabyte errors`".colorize(:green)+"."
  else
    display res.body.to_json
  end
end

#errors ⇒ `Object`

flows:errors ID

Show recent errors generated by the flow.

–output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 294

def errors
  require("colorize")

  # Init
  flow_id = options[:id] || shift_argument

  # No name?
  if flow_id.nil?
    flow_id = read_name_from_conf(options)
  end

  type = options[:output_type]

  # Make the request
  require("cgi")
  res = api.request(
    :expects => 200,
    :method => :get,
    :body    => options.to_json,
    :path => "/flows/#{CGI.escape(flow_id)}/errors"
  )

  # Render
  display "Recent errors:" if type.nil?
  headings = ["operation", "date", "error"]
  rows = (res.body["recent_errors"] || []).map do |row|
    if row['date']
      d = row['date']
    else
      d = nil
    end
    [row['name'], d, row['message']]
  end
  rows.sort! do |a,b|
    a[0] <=> b[0]
  end
  color_map = {}
  colors = [:green, :yellow, :magenta, :cyan, :light_black, :light_green, :light_yellow, :light_blue, :light_magenta, :light_cyan]
  rows.each do |row|
    name = row[0]
    time = row[1]
    message = row[2].strip
    color_map[name] ||= colors.shift
    if time
      display "#{"* #{name} - #{time}".colorize(color_map[name])}:" if type.nil?
    else
      display "#{"* #{name}".colorize(color_map[name])}:" if type.nil?
    end
    message.split('\n').each do |sub_line|
      display "    #{sub_line}" if type.nil?
    end
  end

end

#info ⇒ `Object`

flows:info [DIR]

Outputs the info for the flow in the dir.

–pretty # Pretty prints the info output –dir DIR # Flow directory –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN –host HOST # #HIDDEN –port PORT # #HIDDEN –unix_socket PATH # #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 563

def info
  
  # Init 
  require 'socket'
  type = options[:output_type]
  dir = options[:dir] || shift_argument
  if dir.nil?
    dir = Dir.pwd
  else
    dir = File.expand_path(dir)
  end
  host = options[:host]
  port = options[:port]
  unix_socket = options[:unix_socket]
  
  if (host && port)
    
    # Do we have a host, port? If so, delegate to the multilang...
    flow_info = Zillabyte::Helpers.get_info(dir, options)
    socket = TCPSocket.new(host, port)
    socket.write(flow_info.to_json)
    socket.close()
    exit
    
  elsif unix_socket 
    
    flow_info = Zillabyte::Helpers.get_info(dir, options)
    socket = UNIXSocket.new(unix_socket)
    socket.write(flow_info.to_json)
    socket.close()
    exit
    
  else
    
    # Otherwise, start our own tcp socket and get the info, print to stdout 
    flow_info = Zillabyte::Helpers.get_info(dir, options)
    if options[:pretty]
      puts JSON.pretty_generate(flow_info)
    else
      puts flow_info.to_json
    end
    exit 
    
  end
end

#kill ⇒ `Object`

flows:kill ID

Kills the given flow.

–config CONFIG_FILE # Use the given config file –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 1063

def kill
  id = options[:id] || shift_argument
  type = options[:output_type]

  if id.nil?
    id = read_name_from_conf(options)
  end

  display "Killing flow ##{id}...please wait..." if type.nil?
  response = api.flows.create_kill(id, options)

  if response["job_id"]
    options[:job_id] = response["job_id"]
    flow_id = response["flow_id"]

    start = Time.now.utc
    display "Kill request sent. Please wait..." if type.nil?

    while(Time.now.utc < start + MAX_POLL_SECONDS) do

      # Poll
      res = self.api.flows.poll_kill(flow_id, options)

      case res['status']
      when 'completed'
        if res['return']
          if type == "json"
            display "{}"
          else
            display "Flow ##{id} killed"
          end
        else
          throw "something is wrong: #{res}"
        end
        # success! continue below
        break
      when 'running'
        sleep(POLL_SLEEP)
     #   display ".", false
      else
        throw "unknown status: #{res}"
      end

    end
  elsif response["error"]
    error(response["error"], type)
  else
    error("remote server error (f569)", type)
  end
end

#live_run ⇒ `Object`

flows:live_run [OPERATION_NAME] [PIPE_NAME] [DIR]

Runs a local flow with live data.

–config CONFIG_FILE # Use the given config file –dir DIR # Flow directory –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN –host HOST # #HIDDEN –port PORT # #HIDDEN –unix_socket PATH # #HIDDEN

HIDDEN:

# File 'lib/zillabyte/cli/flows.rb', line 520

def live_run
  require("zillabyte/cli/config")
  name = options[:name] || shift_argument
  type = options[:output_type]

  host = options[:host]
  port = options[:port]
  unix_socket = options[:unix_socket]
  
  dir = options[:dir] || shift_argument
  if dir.nil?
    dir = Dir.pwd
  else
    dir = File.expand_path(dir)
  end

  meta = Zillabyte::CLI::Config.get_config_info(dir, options)

  if meta.nil?
    error("could not find meta information for: #{dir}", type)
  end
 
  if(host && port)
    exec(Zillabyte::Helpers.command("--execute_live --name #{name.to_s} --host \"#{host}\" --port #{port}",type, dir))      
  elsif (unix_socket)
    exec(Zillabyte::Helpers.command("--execute_live --name #{name.to_s} --unix_socket \"#{unix_socket}\"",type, dir))      
  else 
    exec(Zillabyte::Helpers.command("--execute_live --name #{name.to_s}",type, dir))
  end
end

#local ⇒ `Object`

flows:local [RPC_ARGS]

Run a flow locally with sample data.

–dir DIR # Flow directory –input_file INPUT_FILE # Input csv file containing parameters for multiple queries –output_prefix OUTPUT_PREFIX # Prefix for output file containing app results. Results will be stored in [prefix]_.csv for each sink. –env ENV # HIDDEN –use_local_jar # HIDDEN –verbose # HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 703

def local
  require('colorize')
  def on_exit(flow_type)
    display "*"*79
    if flow_type == "component"
      display "If you are happy with your RPC and would like to run it in parallel on our servers, please use `zillabyte execute`.".colorize(:green)
    else
      display "If you are happy with your app and would like to run it in parallel on our servers, please use `zillabyte push`.".colorize(:green)
    end
    display "*"*79
  end

  dir = options[:dir]
  if dir.nil?
    dir = Dir.pwd
  else
    dir = File.expand_path(dir)
  end
  env = options[:env] || "prod"
  
  # Get the flow_type
  require("zillabyte/api/flows")
  meta = get_rich_info(dir, session, {:test => true})
  if meta.nil? || meta["nodes"].nil?
    error "this is not a valid zillabyte flow directory"
    exit
  end

  if meta["flow_type"] == "component" #component
    file = options[:input_file]

    component_input_stream = nil
    fields = []
    meta["nodes"].each do |node|
      if node["type"] == "input"
        component_input_stream = node["name"]
        fields = node["fields"].map {|field| field.keys.first}
        break
      end
    end

    component_args = []
    if file
      require('csv')
      CSV.foreach(file) do |line|
        line_arg = {}
        line.each_with_index do |l, i|
          line_arg[fields[i]] = l
        end
        component_args << line_arg
      end 
    else
      args = {}
      count = 0
      while(true) do
        next_arg = shift_argument
        break if next_arg.nil?
        args[fields[count]] = next_arg
        count += 1
      end
      component_args << args if !args.empty?
    end

    if component_args.empty?
      display "No input arguments given for RPC. Please see 'zillabyte help test' or 'zillabyte help local' for details."
      exit(1)
    end
    component_args = {component_input_stream => component_args}
  else #app
    output_prefix = options[:output_prefix]
  end

  # Check that multilang version is atleast 0.1.0
  version = meta["multilang_version"] || "0.0.0"
  language = meta["language"]
  require('zillabyte/auth')
  auth_token = Zillabyte::Auth.api_key_or_nil
  if !version_okay?(version, Zillabyte::CLI::VERSION)
    display "The version of zillabyte used in your flow is outdated."
    display "Please upgrade your zillabyte CLI using 'bundle update zillabyte; gem cleanup zillabyte'."
    case language
    when "python"
      display "Then upgrade your zillabyte Python library using '[sudo] pip install zillabyte --upgrade'."
    when "javascript"
      # something
    end
    return
  end

  # Take care of downloading the jar
  local = options[:env] == "local" || options[:use_local_jar] ? true : false
  jar_path = Zillabyte::Command::Download.new.jar(local)

  # Submit app to jar
  require('pty')
  require('json')
  require('shellwords')
  cmd = "java -jar #{jar_path} --flow.name #{meta["name"]} --flow.class #{meta["flow_type"]} --directory #{dir}"
  if env == "test" or env == "staging"
    cmd += " --api.host test.api.zillabyte.com --api.port 80"
  elsif env == "prod"
    cmd += " --api.host api.zillabyte.com --api.port 80"
  end 
  cmd += " --rpc.args #{Shellwords.escape(component_args.to_json)}" if component_args
  cmd += " --output.prefix #{output_prefix}" if output_prefix
  cmd += " --auth.token #{auth_token}" if auth_token

  match_str = "[f#{meta["name"]}]"
  display "Begin local execution..."
  begin
    PTY.spawn(cmd) do |r, w, pid|
      done_cycle = false
      begin
        r.each do |line|
          if options[:verbose]
            display line
          else
            if(line.include? match_str)
              line = line.chomp!.split(match_str).last
              if !done_cycle
                if line.include? "[ERROR]"
                  display line.colorize(:red)
                elsif line.include? "sinking tuple:"
                  display line.colorize(:green)
                else
                  display line
                end
              end
            end
            done_cycle = true if(line.include? "[app][RUN] Transitioned to state WAITING" or line.include? "[app][RUN] Transitioned to state IDLE")
          end
        end
      rescue Errno::EIO => e
        # process has finished 
      end
    end
  rescue Exception => e
    # Do nothing on error...
  ensure
    on_exit meta["flow_type"]
  end
end

#logs ⇒ `Object`

flows:logs ID [OPERATION_NAME]

Streams logs for the flow from our cluster.

–operation OPERATION_NAME # Specify the operation to show logs for -v, –verbose LEVEL # Sets the verbosity (error, info, debug) [default: info] –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 264

def logs

  flow_id = options[:id] || shift_argument
  if flow_id.nil?
    flow_id = read_name_from_conf(options)
  end

  operation_id = options[:operation] || shift_argument || '_ALL_'
  category = options[:verbose] || '_ALL_'
  type = options[:output_type]

  carry_settings = {
    :category => category
  }

  display "Retrieving logs for flow ##{flow_id}...please wait..." if type.nil?
  hash = self.api.logs.get(flow_id, operation_id, options)

  fetch_logs(hash, operation_id)

end

#on_exit(flow_type) ⇒ `Object`

# File 'lib/zillabyte/cli/flows.rb', line 705

def on_exit(flow_type)
  display "*"*79
  if flow_type == "component"
    display "If you are happy with your RPC and would like to run it in parallel on our servers, please use `zillabyte execute`.".colorize(:green)
  else
    display "If you are happy with your app and would like to run it in parallel on our servers, please use `zillabyte push`.".colorize(:green)
  end
  display "*"*79
end

#pause ⇒ `Object`

flows:pause ID

Pause a flow

# File 'lib/zillabyte/cli/flows.rb', line 854

def pause
  id = options[:id] || shift_argument
  type = options[:output_type]

  if id.nil?
    id = read_name_from_conf(options)
  end

  display "Pausing flow ##{id}...please wait..." if type.nil?
  response = api.flows.create_pause(id, options)

  if response["job_id"]
    options[:job_id] = response["job_id"]
    flow_id = response["flow_id"]

    start = Time.now.utc
    display "Pause request sent. Please wait..." if type.nil?

    while(Time.now.utc < start + MAX_POLL_SECONDS) do

      # Poll
      res = self.api.flows.poll_pause(flow_id, options)

      case res['status']
      when 'completed'
        if res['return']
          if type == "json"
            display "{}"
          else
            display "Flow ##{id} paused"
          end
        else
          throw "something is wrong: #{res}"
        end
        # success! continue below
        break
      when 'running'
        sleep(POLL_SLEEP)
     #   display ".", false
      else
        throw "unknown status: #{res}"
      end

    end
  elsif response["error"]
    error(response["error"], type)
  else
    error("remote server error (f569)", type)
  end

end

#prep ⇒ `Object`

flows:prep [DIR]

Performs any necessary initialization for the flow.

–dir DIR # Flow directory –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN –mode MODE # Specify development or production mode for dependencies #HIDDEN

# File 'lib/zillabyte/cli/flows.rb', line 206

def prep()
  require("zillabyte/cli/config")
  require("shellwords")
  
  type = options[:output_type]
  dir = options[:dir] || shift_argument
  if dir.nil?
    dir = Dir.pwd
  else
    dir = File.expand_path(dir)
  end
  meta = Zillabyte::CLI::Config.get_config_info(dir)

  if meta.nil?
    error("The specified directory (#{dir}) does not appear to contain a valid Zillabyte configuration file.", type)
  end

  case meta["language"]
  when "ruby"
    
    base_cmd = "cd \"#{meta['home_dir']}\"; unset BUNDLE_GEMFILE; unset RUBYOPT;"
    bundle_install_cmd = case options[:mode]
    when "mock", /^mock-.+/
      # Don't run 'bundle install' in test mode => speed things up.  This assumes
      # all gems are already installed locally!
    else
      exec(base_cmd + "bundle install")
    end
    
  when "python"
    vDir = "#{meta['home_dir']}/vEnv"
    lock_file = meta['home_dir']+"/zillabyte_thread_lock_file"
    if File.exists?(lock_file)
      sleep(1) while File.exists?(lock_file)
    else
      begin
        # TODO: won't the ~/zb1/ break in LXC and users' machines? 
        cmd = "touch #{Shellwords.escape(lock_file)}; virtualenv --clear --system-site-packages #{Shellwords.escape(vDir)}; PYTHONPATH=$PYTHONPATH:~/zb1/multilang/python/Zillabyte #{Shellwords.escape(vDir)}/bin/pip install -r #{Shellwords.escape(File.join(meta['home_dir'], "requirements.txt"))}"
        system cmd, :out => :out
      ensure
        File.delete(lock_file) if File.exists?(lock_file)
      end
    end
    
  when "js"
  end
  
end

#pull ⇒ `Object`

flows:pull ID DIR

Pulls a flow source to a directory.

–force # Pulls even if the directory exists –dir DIR # Flow directory –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN

Examples:

$ zillabyte flows:pull 27 .

# File 'lib/zillabyte/cli/flows.rb', line 22

def pull

  flow_id = options[:id] || shift_argument
  type = options[:output_type]

  dir = options[:dir] || shift_argument
  error("no directory given", type) if dir.nil?
  dir = File.expand_path(dir)
  
  error("no id given", type) if flow_id.nil?
  
  # Create if not exists..
  if File.exists?(dir)
    if Dir.entries(dir).size != 2 and options[:force].nil?
      error("target directory not empty. use --force to override", type)
    end
  else
    require("fileutils")
    FileUtils.mkdir_p(dir)
  end
  
  res = api.flows.pull_to_directory flow_id, dir, session, options
  
  if res['error']
    error("error: #{res['error_message']}", type)
  else
    if type == "json"
      display "{}"
    else
      display "Flow ##{res['id']} pulled to #{friendly_dir(dir)}"
    end
  end
  
end

#push ⇒ `Object`

flows:push [DIR]

Uploads a flow. –force # Don’t ask for confirmation –config CONFIG_FILE # Use the given config file –dir DIR # Flow directory –output_type OUTPUT_TYPE # Specify an output type i.e. json #HIDDEN –public # Make the flow public

# File 'lib/zillabyte/cli/flows.rb', line 67

def push
  since = Time.now.utc.to_s
  dir = options[:dir]
  if dir.nil?
    dir = Dir.pwd
  else
    dir = File.expand_path(dir)
  end
  type = options[:output_type]
  forced = options[:force]

  # Get the meta info
  require('zillabyte/api/flows')
  options[:local_flag] = 1
  hash = get_rich_info(dir, session, options)
  if hash.nil?
    session.error("unable to extract #{flow_type} meta information", type) if session
  end

  if hash["error"]
    session.error(hash["error_message"], type) if session
  end

  flow_type = hash["flow_type"]
  flow_name = hash["name"]

  # Before killing live flow (with same name), ask for confirmation before push unless force flag is on 
  if !forced
    if flow_type == "app"
      flows = api.app.list
    else
      flows = api.component.list
    end
    flows.each do |flow|
      if flow_name == flow["name"]
        if flow["state"] == "RUNNING"
          
          display "This flow is already running and will be terminated. Continue? [y/N]: ", false
          input = ask          
          if !(input.downcase == "y" || input.downcase == "yes")
            return
          end
        end
        break
      end
    end
  end 
  
  res = api.flows.push_directory dir, hash, session, options

  if res['error']
    error("error: #{res['error_message']}", type)
  else
    display "#{flow_type} ##{res['id']} #{res['action']}" if type.nil?
  end

  display "Setting up your #{flow_type}... please wait..." if type.nil?
  sleep(2) # wait for kill command 

  hash = self.api.logs.get(res['id'], nil, options)
  if flow_type == "app"
    fetch_logs(hash, "_ALL_", ["App deployed", "STOP_LOGGING"])
  elsif flow_type == "component"
    fetch_logs(hash, "_ALL_", ["Component registered", "STOP_LOGGING"])
  end
  
  if type == "json"
    display "{}" 
  else
    display "*"*79
    if flow_type == "app"
      display "App deployed: #{get_flow_ui_link(res).colorize(:light_yellow)}"
      display "`zillabyte details`".colorize(:green)+"  :   to check the progress of your app."
      display "`zillabyte logs`".colorize(:green)+"     :   to look at log output."
      display "`zillabyte kill`".colorize(:green)+"     :   to stop your app on our servers."
      display "To inspect/download data produced by your app:"
      display "`zillabyte data:show [RELATION_NAME]`".colorize(:cyan)+"                              :   to inspect your data."
      display "`zillabyte data:pull [RELATION_NAME] [OUTPUT_PREFIX] [DIRECTORY]`".colorize(:cyan)+"  :   to download your data."
    elsif flow_type == "component"
      display "Component registered: #{get_flow_ui_link(res).colorize(:light_yellow)}"
      display "Your component may now be embedded in any application!"
      display "`zillabyte execute`".colorize(:green)+"  :   to run your component as an RPC."
    end
    display "*"*79
  end

end

#resume ⇒ `Object`

flows:resume

Resume a paused flow ID

# File 'lib/zillabyte/cli/flows.rb', line 910

def resume
  id = options[:id] || shift_argument
  type = options[:output_type]

  if id.nil?
    id = read_name_from_conf(options)
  end

  display "Resume flow ##{id}...please wait..." if type.nil?
  response = api.flows.create_resume(id, options)

  if response["job_id"]
    options[:job_id] = response["job_id"]
    flow_id = response["flow_id"]

    start = Time.now.utc
    display "Resume request sent. Please wait..." if type.nil?

    while(Time.now.utc < start + MAX_POLL_SECONDS) do

      # Poll
      res = self.api.flows.poll_resume(flow_id, options)

      case res['status']
      when 'completed'
        if res['return']
          if type == "json"
            display "{}"
          else
            display "Flow ##{id} resumed"
          end
        else
          throw "something is wrong: #{res}"
        end
        # success! continue below
        break
      when 'running'
        sleep(POLL_SLEEP)
     #   display ".", false
      else
        throw "unknown status: #{res}"
      end

    end
  elsif response["error"]
    error(response["error"], type)
  else
    error("remote server error (f569)", type)
  end
end

#scale ⇒ `Object`

apps:scale ID OPERATION_NAME PARALLELISM

Change the parallelism of of app or operation

# File 'lib/zillabyte/cli/flows.rb', line 969

def scale
  id = options[:id] || shift_argument

  if id.nil?
    id = read_name_from_conf(options)
  end

  name = options[:name] || shift_argument
  parallelism = options[:parallelism] || shift_argument
  type = options[:output_type]

  # Read current metadata from API
  display "Fetching configuration for flow ##{id}...please wait..." if type.nil?
  options[:flow_id] = id

  response = api.flows.get_meta(id, options)
  if response["meta"]
    meta = response["meta"]
    nodes = meta["nodes"]

    # Set new parallelism if necessary
    scaling = false
    nodes.each_with_index do |node, i|

      if node["name"] == name && node["parallelism"] != parallelism
        meta["nodes"][i]["parallelism"] = parallelism
        scaling = true
        break
      end
    end

    if scaling
      display "Scaling flow ##{id}...please wait..." if type.nil?
      options[:flow_config] = meta
      response = api.flows.create_scale(id, options)

      if response["job_id"]
        options[:job_id] = response["job_id"]
        flow_id = response["flow_id"]

        start = Time.now.utc
        display "Scaling request sent. Please wait..." if type.nil?

        while(Time.now.utc < start + MAX_POLL_SECONDS) do

          # Poll
          options.delete(:flow_config)
          res = self.api.flows.poll_scale(flow_id, options)

          case res['status']
          when 'completed'
            if res['return']
              if type == "json"
                display "{}"
              else
                display "Flow ##{id} scaled"
              end
            else
              throw "something is wrong: #{res}"
            end
            # success! continue below
            break
          when 'running'
            sleep(POLL_SLEEP)
         #   display ".", false
          else
            throw "unknown status: #{res}"
          end

        end
      elsif response["error"]
        error(response["error"], type)
      else
        error("remote server error (f569)", type)
      end
    else
      error("Unable to find operation #{name}", type)
    end
  elsif response["error"]
    error(response["error"], type)
  end

end

Class: Zillabyte::Command::Flows

Overview

Direct Known Subclasses

Constant Summary collapse

Constants inherited from Base

Instance Attribute Summary

Attributes inherited from Base

Instance Method Summary collapse

Methods inherited from Base

Methods included from Helpers

Constructor Details

Instance Method Details

#authorize ⇒ Object

#delete ⇒ Object

#details ⇒ Object

#errors ⇒ Object

#info ⇒ Object

#kill ⇒ Object

#live_run ⇒ Object

#local ⇒ Object

#logs ⇒ Object

#on_exit(flow_type) ⇒ Object

#pause ⇒ Object

#prep ⇒ Object

#pull ⇒ Object

#push ⇒ Object

#resume ⇒ Object

#scale ⇒ Object

#authorize ⇒ `Object`