package pl.lusch.elastic.dc.collectors;

import java.util.Iterator;
import java.util.concurrent.ScheduledFuture;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsRequest;
import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.rest.RestController;
import org.json.JSONArray;
import org.json.JSONObject;

import pl.lusch.elastic.dc.scheduler.EveryFiveSeconds;

public class StatsCollector extends BaseCollector {
	
	private final String indexType = "stats";
	
    @Inject	
    public StatsCollector(Settings settings, RestController controller, Client client) {
		super(settings, client);
	}

	@Override
	protected void getCollectorProperties() {
		// TODO get properties
	}

	@Override
	public void sheduleCollectionJob() {
		
		// Collection job
		final Runnable job = new Runnable() {
			public void run() {
				
				// The collector job

				// Prepare and configure statistics request to ES
				NodesStatsRequest nodesStatsRequest = new NodesStatsRequest();
				nodesStatsRequest.indices(false).os(true).fs(true).threadPool(true);
				
				// Request for stats
				clusterClient.admin().cluster().nodesStats(nodesStatsRequest, new ActionListener<NodesStatsResponse>() {

					public void onResponse(NodesStatsResponse response) {

						// Convert response string to JSONArray
						JSONArray responseArray = new JSONArray("["+response+"]");
						JSONObject responseObject = responseArray.getJSONObject(0);
						
						// Get all cluster nodes
						JSONObject nodes = responseObject.getJSONObject("nodes");

						@SuppressWarnings("unchecked")
						Iterator<String> itr = nodes.keys();
						
						while(itr.hasNext()) {
							String nodeId = itr.next();
							
							// Get node object
							JSONObject node = (JSONObject) nodes.get(nodeId);
							String nodeName = node.getString("name");
							
							// Index node statistics back to ES
							IndexResponse re = clusterClient.prepareIndex("nodes", indexType)
								.setSource(node.toString())
								.execute()
								.actionGet();
							
							logger.info(
								"Collected stats at: " + re.getIndex() + "/" + re.getType() + "/" + re.getId() + "."
							);
						}
					}

					public void onFailure(Throwable e) {
						logger.error(e.getMessage(), e.getCause());
					}
					
				});
				
			}
		};
		
		// run job every 5 seconds
		ScheduledFuture<?> jobHandle = (new EveryFiveSeconds(job)).getJobHandle();
		
		if(jobHandle.isDone()) {
			logger.info("The job is done");
		}
	}

}

@@ -711,16 +711,59 @@
 tor %7B%0A%09%0A
+%09private final String indexName = %22nodes%22;%0A
 %09private
@@ -797,16 +797,16 @@
 stats%22;%0A
-
 %09%0A    @I
@@ -928,16 +928,473 @@
 lient);%0A
+%09%09%0A%09%09curl -XDELETE 'localhost:9200/nodes?pretty'%0A%09%09/*%0A%09%09curl -XPOST 'localhost:9200/nodes?pretty' -d '%7B%0A%09%09  %22mappings%22: %7B%0A%09%09    %22stats%22: %7B%0A%09%09      %22properties%22: %7B%0A%09%09        %22timestamp%22: %7B%0A%09%09          %22type%22: %22date%22, %22index%22: %22not_analyzed%22%0A%09%09        %7D,%0A%09%09        %22host%22: %7B%0A%09%09          %22type%22: %22string%22, %22index%22: %22not_analyzed%22%0A%09%09        %7D,%0A%09%09        %22name%22: %7B%0A%09%09          %22type%22: %22string%22, %22index%22: %22not_analyzed%22%0A%09%09        %7D%0A%09%09      %7D%0A%09%09    %7D%0A%09%09  %7D%0A%09%09%7D'*/%0A
 %09%7D%0A%0A%09@Ov
@@ -1805,16 +1805,26 @@
 s(true).
+jvm(true).
 fs(true)
@@ -2717,23 +2717,25 @@
 reIndex(
-%22nodes%22
+indexName
 , indexT

enable jvm stats