STORM-200 Proposal for Multilang's Metrics feature

for https://issues.apache.org/jira/browse/STORM-200

Storm 0.9.2 exposes a metrics interface to report summary statistics across the full topology. We can build our own metric, and build metrics consumer to use those statistics. 

But when we use Multilang(ie. Python), we can not use this feature. So we want to summit a proposal for multilang's metrics. 

--------------------------------------------------------------------------------------------------------------------------------------
The specifics of the proposal: 

1. The main idea is: when user want to add a metric statistics in multilang(python) spout/bolt， 
    a) first, create a metric object and register in ShellSpout/ShellBolt's sub-class, 
    b) then update the metric in Python spout/bolt process through RPC call. 

2. Extends Metrics API with RPC: 
    In package backtype.storm.metric.api add new package backtype.storm.metric.api.rpc, which includes:
    a) IShellMetric interface, which extends IMetirc interface, to support: 
                   public void updateMetricFromRPC(Object params); 
    b) IShellMetric implemention: AssignableShellMetric, CombinedShellMetric, CountShellMetirc, ReducedShellMetric; 

3. Extends Multilang protocol:
    a) add a command "metrics" for shell process to make RPC call. The protocol is: 
         {"command":"metrics", "name":"metric-registerd-name", "params": param-value} 
    b) moidfy ShellSpout/ShellBolt to support command metrics: we define a method handleMetrics to handle RPC call from Python spout/bolt process； 
    c) modify TopologyContext.java, add getRegisteredMetricByName to get registerd metric to update;
    c) storm.py add method rpcMetrics(name, params), user can update remote metric through this RPC call. 

4. Extends TopologyContext add getRegisteredMetricByName to get a component's registered metric to update through rpc;

--------usage example----------------------------------------------------------------------------------------------------------------
--------ExamplePythonBolt.java----------------------------------------------------------------------------------------------------

    public class ExamplePythonBolt extends ShellBolt implements IRichBolt {

	public ExamplePythonBolt() {
		super("python", "ExampleBolt.py");
	}
	
       public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
           super.prepare(stormConf, context, collector);
    	
           //generate a metric object
    	   CountShellMetric cMetric = new CountShellMetric(); 
           //register metric object to context             
    	   context.registerMetric("PythonBoltCount", cMetric, 120);  
       }

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("LogRecord"));
	}

	public Map<String, Object> getComponentConfiguration() {
		return null;
	}
    }

--------ExampleBolt.py---------------------------------------------------------------------------------------------------

    class ExampleBolt(storm.BasicBolt):

        def __init__(self, boltParams):
            #from here, you can get params from java
            pass
    
        def initialize(self, stormconf, context):
            pass
        
        def process(self, tup):
            try:         
                storm.emit([tup[0]])
            
                 //update your PythonBoltCount metric with param 1
                storm.rpcMetrics("PythonBoltCount", 1)       
            except Exception,tx:
                //handle error

--------ExamplePythonSpout.java---------------------------------------------------------------------------------------------------
	
        public class ExamplePythonSpout extends ShellSpout implements IRichSpout {
		public ExamplePythonSpout() {
			super("python", "ExampleSpout.py");
		}
		
		@Override
		public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
			super.open(conf, context, collector);
		
                        //generate a metric object
			MTCountMetric cMetric = new MTCountMetric();
                        //register metric object to context   
			context.registerMetric("PythonSpoutCount", cMetric, 120);
		}
	
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
			declarer.declare(new Fields("str"));
		}
	
		public Map<String, Object> getComponentConfiguration() {
			return null;
		}
	}

--------ExampleSpout.py---------------------------------------------------------------------------------------------------

    class ExampleSpout(storm.Spout):
        def __init__(self):
            pass
        
        def initialize(self, stormconf, context):
            pass
        
        def ack(self, id):
            pass
        
        def fail(self, id):
            pass
            
        def nextTuple(self):
            str = "this is a spout message generated by python"
            storm.emit([str])
            
            //update your PythonBoltCount metric with param 1
            storm.rpcMetrics("PythonSpoutCount", 1)
Sleep in a test to get the timing right is not Ideal.  I can see in the other multi-lang tests they mostly just verify that the topology ran, nothing about the metrics being consumed.  I would prefer to see you use with-simulated-time-local-cluster instead.  I don't see any reason why they should not work with shell bolts and spouts.