core/src/main/java/org/apache/hadoop/chukwa/datatrigger/HttpTriggerAction.java - chukwa - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.chukwa.datatrigger;

 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.BufferedReader;
 import java.io.OutputStreamWriter;
 import java.net.URL;
 import java.net.HttpURLConnection;
 import java.net.MalformedURLException;
 import java.nio.charset.Charset;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.HashMap;
 import java.util.Map.Entry;

 /**
  * Trigger action that makes an HTTP request when executed.
  * <P>
  * To use this trigger, two types of configurations must be set. First, this class
  * must be configured to be invoked for a given trigger event. Second, the
  * the relevant settings for the HTTP request(s) to be made must be set as
  * described below.
  * <P>
  * The general format of this classes configs is
  * <code>chukwa.trigger.[eventName].http.[N].[paramName]</code> where
  * <code>eventName</code> is the name of the event the request values are bound
  * to (see TriggerEvent), <code>N</code> is a counter for each request configured (starting at 1)
  * and <code>paramName</code> is the request parameter being set.
  * <P>
  * Using the post demux success trigger event as an example, the first request
  * to be fired would use the following configurations
  * <ul>
  * <li><code>chukwa.trigger.post.demux.success.http.1.url</code> - The HTTP url to
  * invoke.</li>
  * <li><code>chukwa.trigger.post.demux.success.http.1.method</code> - The HTTP method
  * (optional, default=GET).</li>
  * <li><code>chukwa.trigger.post.demux.success.http.1.headers</code> - A comma-delimited
  * set of HTTP headers (in <code>[headerName]:[headerValue]</code> form) to
  * include (optional).</li>
  * <li><code>chukwa.trigger.post.demux.success.http.1.body</code> - The text HTTP body
  * to include (optional).</li>
  * <li><code>chukwa.trigger.post.demux.success.http.1.connect.timeout</code> - The
  * HTTP connection timeout setting in milliseconds (optional, default=5000ms).</li>
  * <li><code>chukwa.trigger.post.demux.success.http.1.read.timeout</code> - The
  * HTTP read timeout setting in milliseconds (optional, default=5000ms).</li>
  * </ul>
  * @see TriggerAction
  * @see TriggerEvent
  */
 public class HttpTriggerAction implements TriggerAction {
   protected Log log = LogFactory.getLog(getClass());


   /**
    * Iterates over each URL found, fetched other settings and fires and HTTP
    * request.
    *
    * @param conf is Chukwa configuration
    * @param fs is HDFS File System
    * @param src is list of sources to look for data
    * @param triggerEvent is type of processing to happen
    * @throws IOException if error in process triggers
    */
   public void execute(Configuration conf, FileSystem fs,
                       FileStatus[] src, TriggerEvent triggerEvent) throws IOException {

     if (log.isDebugEnabled()) {
       for (FileStatus file : src) {
           log.debug("Execute file: " + file.getPath());
       }
     }

     int reqNumber = 1;
     URL url = null;
     while ((url = getUrl(conf, triggerEvent, reqNumber)) != null) {

       // get settings for this request
       String method = getMethod(conf, triggerEvent, reqNumber);
       Map<String, String> headers = getHeaders(conf, triggerEvent, reqNumber);
       String body = getBody(conf, triggerEvent, reqNumber);
       int connectTimeout = getConnectTimeout(conf, triggerEvent, reqNumber);
       int readTimeout = getReadTimeout(conf, triggerEvent, reqNumber);

       try {
         // make the request
         makeHttpRequest(url, method, headers, body, connectTimeout, readTimeout);
       }
       catch(Exception e) {
         log.error("Error making request to " + url, e);
       }
       reqNumber++;
     }
   }

   private void makeHttpRequest(URL url, String method,
                                Map<String, String> headers, String body,
                                int connectTimeout, int readTimeout) throws IOException {
     if (url == null) {
       return;
     }

     // initialize the connection
     HttpURLConnection conn = (HttpURLConnection)url.openConnection();
     conn.setRequestMethod(method);
     conn.setDoInput(true);
     conn.setConnectTimeout(connectTimeout);
     conn.setReadTimeout(readTimeout);

     // set headers
     boolean contentLengthExists = false;
     if (headers != null) {
       for(Entry<String, String> entry : headers.entrySet()) {
         String name = entry.getKey();
         String value = entry.getValue();
         if (log.isDebugEnabled()) {
           log.debug("Setting header " + name + ": " + value);
         }
         if (name.equalsIgnoreCase("content-length")) {
           contentLengthExists = true;
         }
         conn.setRequestProperty(name, value);
       }
     }

     // set content-length if not already set
     if (!"GET".equals(method) && !contentLengthExists) {
       String contentLength = body != null ? String.valueOf(body.length()) : "0";
       conn.setRequestProperty("Content-Length", contentLength);
     }

     // send body if it exists
     if (body != null) {
       conn.setDoOutput(true);
       OutputStreamWriter writer = new OutputStreamWriter(conn.getOutputStream(), Charset.forName("UTF-8"));
       writer.write(body);
       writer.flush();
       writer.close();
     }
     else {
       conn.setDoOutput(false);
     }

     // read reponse code/message and dump response
     log.info("Making HTTP " + method + " to: " + url);
     int responseCode = conn.getResponseCode();
     log.info("HTTP Response code: " + responseCode);


     if (responseCode != 200) {
       log.info("HTTP Response message: " + conn.getResponseMessage());
     }
     else {
       BufferedReader reader = new BufferedReader(
                                 new InputStreamReader(conn.getInputStream(), Charset.forName("UTF-8")));
       String line;
       StringBuilder sb = new StringBuilder();
       while ((line = reader.readLine()) != null) {
         if(sb.length() > 0) {
           sb.append("\n");
         }
         sb.append(line);
       }
       log.info("HTTP Response:\n" + sb);

       reader.close();
     }

     conn.disconnect();
   }

   protected URL getUrl(Configuration conf,
                        TriggerEvent triggerEvent,
                        int reqNumber) throws MalformedURLException {
     String urlString = conf.get(getConfigKey(triggerEvent, reqNumber, "url"), null);
     if (urlString == null) {
       return null;
     }

     return new URL(urlString);
   }

   protected String getMethod(Configuration conf,
                              TriggerEvent triggerEvent,
                              int reqNumber) {
     return conf.get(getConfigKey(triggerEvent, reqNumber, "method"), "GET");
   }

   protected Map<String, String> getHeaders(Configuration conf,
                                            TriggerEvent triggerEvent,
                                            int reqNumber) {
     Map<String, String> headerMap = new HashMap<String,String>();

     String headers = conf.get(getConfigKey(triggerEvent, reqNumber, "headers"), null);

     if (headers != null) {
       String[] headersSplit = headers.split(",");
       for (String header : headersSplit) {
         String[] nvp = header.split(":", 2);
         if (nvp.length < 2) {
           log.error("Invalid HTTP header found: " + Arrays.toString(nvp));
           continue;
         }
         headerMap.put(nvp[0].trim(), nvp[1].trim());
       }
     }

     return headerMap;
   }

   protected String getBody(Configuration conf,
                            TriggerEvent triggerEvent,
                            int reqNumber) {
     return conf.get(getConfigKey(triggerEvent, reqNumber, "body"), "GET");
   }

   protected int getConnectTimeout(Configuration conf,
                                  TriggerEvent triggerEvent,
                                  int reqNumber) {
     String timeout = conf.get(getConfigKey(triggerEvent, reqNumber, "connect.timeout"), null);
     return timeout != null ? Integer.parseInt(timeout) : 5000;
   }


   protected int getReadTimeout(Configuration conf,
                               TriggerEvent triggerEvent,
                               int reqNumber) {
     String timeout = conf.get(getConfigKey(triggerEvent, reqNumber, "read.timeout"), null);
     return timeout != null ? Integer.parseInt(timeout) : 5000;
   }

   private String getConfigKey(TriggerEvent triggerEvent, int reqNumber, String name) {
     return triggerEvent.getConfigKeyBase() + ".http." + reqNumber + "." + name;
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.chukwa.datatrigger;

	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;

	import java.io.IOException;
	import java.io.InputStreamReader;
	import java.io.BufferedReader;
	import java.io.OutputStreamWriter;
	import java.net.URL;
	import java.net.HttpURLConnection;
	import java.net.MalformedURLException;
	import java.nio.charset.Charset;
	import java.util.Arrays;
	import java.util.Map;
	import java.util.HashMap;
	import java.util.Map.Entry;

	/**
	* Trigger action that makes an HTTP request when executed.
	* <P>
	* To use this trigger, two types of configurations must be set. First, this class
	* must be configured to be invoked for a given trigger event. Second, the
	* the relevant settings for the HTTP request(s) to be made must be set as
	* described below.
	* <P>
	* The general format of this classes configs is
	* <code>chukwa.trigger.[eventName].http.[N].[paramName]</code> where
	* <code>eventName</code> is the name of the event the request values are bound
	* to (see TriggerEvent), <code>N</code> is a counter for each request configured (starting at 1)
	* and <code>paramName</code> is the request parameter being set.
	* <P>
	* Using the post demux success trigger event as an example, the first request
	* to be fired would use the following configurations
	* <ul>
	* <li><code>chukwa.trigger.post.demux.success.http.1.url</code> - The HTTP url to
	* invoke.</li>
	* <li><code>chukwa.trigger.post.demux.success.http.1.method</code> - The HTTP method
	* (optional, default=GET).</li>
	* <li><code>chukwa.trigger.post.demux.success.http.1.headers</code> - A comma-delimited
	* set of HTTP headers (in <code>[headerName]:[headerValue]</code> form) to
	* include (optional).</li>
	* <li><code>chukwa.trigger.post.demux.success.http.1.body</code> - The text HTTP body
	* to include (optional).</li>
	* <li><code>chukwa.trigger.post.demux.success.http.1.connect.timeout</code> - The
	* HTTP connection timeout setting in milliseconds (optional, default=5000ms).</li>
	* <li><code>chukwa.trigger.post.demux.success.http.1.read.timeout</code> - The
	* HTTP read timeout setting in milliseconds (optional, default=5000ms).</li>
	* </ul>
	* @see TriggerAction
	* @see TriggerEvent
	*/
	public class HttpTriggerAction implements TriggerAction {
	protected Log log = LogFactory.getLog(getClass());


	/**
	* Iterates over each URL found, fetched other settings and fires and HTTP
	* request.
	*
	* @param conf is Chukwa configuration
	* @param fs is HDFS File System
	* @param src is list of sources to look for data
	* @param triggerEvent is type of processing to happen
	* @throws IOException if error in process triggers
	*/
	public void execute(Configuration conf, FileSystem fs,
	FileStatus[] src, TriggerEvent triggerEvent) throws IOException {

	if (log.isDebugEnabled()) {
	for (FileStatus file : src) {
	log.debug("Execute file: " + file.getPath());
	}
	}

	int reqNumber = 1;
	URL url = null;
	while ((url = getUrl(conf, triggerEvent, reqNumber)) != null) {

	// get settings for this request
	String method = getMethod(conf, triggerEvent, reqNumber);
	Map<String, String> headers = getHeaders(conf, triggerEvent, reqNumber);
	String body = getBody(conf, triggerEvent, reqNumber);
	int connectTimeout = getConnectTimeout(conf, triggerEvent, reqNumber);
	int readTimeout = getReadTimeout(conf, triggerEvent, reqNumber);

	try {
	// make the request
	makeHttpRequest(url, method, headers, body, connectTimeout, readTimeout);
	}
	catch(Exception e) {
	log.error("Error making request to " + url, e);
	}
	reqNumber++;
	}
	}

	private void makeHttpRequest(URL url, String method,
	Map<String, String> headers, String body,
	int connectTimeout, int readTimeout) throws IOException {
	if (url == null) {
	return;
	}

	// initialize the connection
	HttpURLConnection conn = (HttpURLConnection)url.openConnection();
	conn.setRequestMethod(method);
	conn.setDoInput(true);
	conn.setConnectTimeout(connectTimeout);
	conn.setReadTimeout(readTimeout);

	// set headers
	boolean contentLengthExists = false;
	if (headers != null) {
	for(Entry<String, String> entry : headers.entrySet()) {
	String name = entry.getKey();
	String value = entry.getValue();
	if (log.isDebugEnabled()) {
	log.debug("Setting header " + name + ": " + value);
	}
	if (name.equalsIgnoreCase("content-length")) {
	contentLengthExists = true;
	}
	conn.setRequestProperty(name, value);
	}
	}

	// set content-length if not already set
	if (!"GET".equals(method) && !contentLengthExists) {
	String contentLength = body != null ? String.valueOf(body.length()) : "0";
	conn.setRequestProperty("Content-Length", contentLength);
	}

	// send body if it exists
	if (body != null) {
	conn.setDoOutput(true);
	OutputStreamWriter writer = new OutputStreamWriter(conn.getOutputStream(), Charset.forName("UTF-8"));
	writer.write(body);
	writer.flush();
	writer.close();
	}
	else {
	conn.setDoOutput(false);
	}

	// read reponse code/message and dump response
	log.info("Making HTTP " + method + " to: " + url);
	int responseCode = conn.getResponseCode();
	log.info("HTTP Response code: " + responseCode);


	if (responseCode != 200) {
	log.info("HTTP Response message: " + conn.getResponseMessage());
	}
	else {
	BufferedReader reader = new BufferedReader(
	new InputStreamReader(conn.getInputStream(), Charset.forName("UTF-8")));
	String line;
	StringBuilder sb = new StringBuilder();
	while ((line = reader.readLine()) != null) {
	if(sb.length() > 0) {
	sb.append("\n");
	}
	sb.append(line);
	}
	log.info("HTTP Response:\n" + sb);

	reader.close();
	}

	conn.disconnect();
	}

	protected URL getUrl(Configuration conf,
	TriggerEvent triggerEvent,
	int reqNumber) throws MalformedURLException {
	String urlString = conf.get(getConfigKey(triggerEvent, reqNumber, "url"), null);
	if (urlString == null) {
	return null;
	}

	return new URL(urlString);
	}

	protected String getMethod(Configuration conf,
	TriggerEvent triggerEvent,
	int reqNumber) {
	return conf.get(getConfigKey(triggerEvent, reqNumber, "method"), "GET");
	}

	protected Map<String, String> getHeaders(Configuration conf,
	TriggerEvent triggerEvent,
	int reqNumber) {
	Map<String, String> headerMap = new HashMap<String,String>();

	String headers = conf.get(getConfigKey(triggerEvent, reqNumber, "headers"), null);

	if (headers != null) {
	String[] headersSplit = headers.split(",");
	for (String header : headersSplit) {
	String[] nvp = header.split(":", 2);
	if (nvp.length < 2) {
	log.error("Invalid HTTP header found: " + Arrays.toString(nvp));
	continue;
	}
	headerMap.put(nvp[0].trim(), nvp[1].trim());
	}
	}

	return headerMap;
	}

	protected String getBody(Configuration conf,
	TriggerEvent triggerEvent,
	int reqNumber) {
	return conf.get(getConfigKey(triggerEvent, reqNumber, "body"), "GET");
	}

	protected int getConnectTimeout(Configuration conf,
	TriggerEvent triggerEvent,
	int reqNumber) {
	String timeout = conf.get(getConfigKey(triggerEvent, reqNumber, "connect.timeout"), null);
	return timeout != null ? Integer.parseInt(timeout) : 5000;
	}


	protected int getReadTimeout(Configuration conf,
	TriggerEvent triggerEvent,
	int reqNumber) {
	String timeout = conf.get(getConfigKey(triggerEvent, reqNumber, "read.timeout"), null);
	return timeout != null ? Integer.parseInt(timeout) : 5000;
	}

	private String getConfigKey(TriggerEvent triggerEvent, int reqNumber, String name) {
	return triggerEvent.getConfigKeyBase() + ".http." + reqNumber + "." + name;
	}
	}