src/storm.thrift

#!/usr/local/bin/thrift --gen java:beans,nocamel,hashcode

namespace java backtype.storm.generated

struct NullStruct {
  
}

union Grouping {
  1: list<string> fields; //empty list means global grouping
  2: NullStruct shuffle; // tuple is sent to random task
  3: NullStruct all; // tuple is sent to every task
  4: NullStruct none; // tuple is sent to a single task (storm's choice) -> allows storm to optimize the topology by bundling tasks into a single process
  5: NullStruct direct; // this bolt expects the source bolt to send tuples directly to it
}

struct StreamInfo {
  1: required list<string> output_fields;
  2: required bool direct;
}

struct ShellComponent {
  1: string execution_command;
  2: string script;
}

union ComponentObject {
  1: binary serialized_java;
  2: ShellComponent shell;
}

struct ComponentCommon {
  1: required map<i32, StreamInfo> streams; //key is stream id
  2: optional i32 parallelism_hint; //how many threads across the cluster should be dedicated to this component
}

struct SpoutSpec {
  1: required ComponentObject spout_object;
  2: required ComponentCommon common;
  3: required bool distributed;
}

struct GlobalStreamId {
  1: required i32 componentId;
  2: required i32 streamId;
  #Going to need to add an enum for the stream type (NORMAL or FAILURE)
}

struct Bolt {
  1: required map<GlobalStreamId, Grouping> inputs; //a join would have multiple inputs
  2: required ComponentObject bolt_object;
  3: required ComponentCommon common;
}

// not implemented yet
// this will eventually be the basis for subscription implementation in storm
struct StateSpoutSpec {
  1: required ComponentObject state_spout_object;
  2: required ComponentCommon common;
}

struct StormTopology {
  //ids must be unique across maps
  1: required map<i32, SpoutSpec> spouts;
  2: required map<i32, Bolt> bolts;
  3: required map<i32, StateSpoutSpec> state_spouts;
  // #workers to use is in conf
}

exception AlreadyAliveException {
  1: required string msg;
}

exception NotAliveException {
  1: required string msg;
}

exception InvalidTopologyException {
  1: required string msg;
}

struct TopologySummary {
  1: required string id;
  2: required string name;
  3: required i32 num_tasks;
  4: required i32 num_workers;
  5: required i32 uptime_secs;
}

struct SupervisorSummary {
  1: required string host;
  2: required i32 uptime_secs;
  3: required i32 num_workers;
  4: required i32 num_used_workers;  
}

struct ClusterSummary {
  1: required list<SupervisorSummary> supervisors;
  2: required i32 nimbus_uptime_secs;
  3: required list<TopologySummary> topologies;
}

struct ErrorInfo {
  1: required string error;
  2: required i32 error_time_secs;
}

struct BoltStats {
  1: required map<string, map<GlobalStreamId, i64>> acked;  
  2: required map<string, map<GlobalStreamId, i64>> failed;  
  3: required map<string, map<GlobalStreamId, double>> process_ms_avg;
}

struct SpoutStats {
  1: required map<string, map<i32, i64>> acked;
  2: required map<string, map<i32, i64>> failed;
  3: required map<string, map<i32, double>> complete_ms_avg;
}

union TaskSpecificStats {
  1: BoltStats bolt;
  2: SpoutStats spout;
}

// Stats are a map from the time window (all time or a number indicating number of seconds in the window)
//    to the stats. Usually stats are a stream id to a count or average.
struct TaskStats {
  1: required map<string, map<i32, i64>> emitted;
  2: required map<string, map<i32, i64>> transferred;
  3: required TaskSpecificStats specific;
}

struct TaskSummary {
  1: required i32 task_id;
  2: required i32 component_id;
  3: required string host;
  4: required i32 port;
  5: required i32 uptime_secs;
  6: required list<ErrorInfo> errors;
  7: optional TaskStats stats;
}

struct TopologyInfo {
  1: required string id;
  2: required string name;
  3: required i32 uptime_secs;
  4: required list<TaskSummary> tasks;
}

service Nimbus {
  void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology) throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite);
  void killTopology(1: string name) throws (1: NotAliveException e);
  // need to add functions for asking about status of storms, what nodes they're running on, looking at task logs

  string beginFileUpload();
  void uploadChunk(1: string location, 2: binary chunk);
  void finishFileUpload(1: string location);
  
  string beginFileDownload(1: string file);
  //can stop downloading chunks when receive 0-length byte array back
  binary downloadChunk(1: string id);
  
  // stats functions
  ClusterSummary getClusterInfo();
  TopologyInfo getTopologyInfo(1: string id) throws (1: NotAliveException e);
  //returns json
  string getTopologyConf(1: string id) throws (1: NotAliveException e);
  StormTopology getTopology(1: string id) throws (1: NotAliveException e);
}

service DistributedRPC {
  string execute(1: string functionName, 2: string funcArgs);
  void result(1: string id, 2: string result);
}