Moved Flume Http Source here
diff --git a/pom.xml b/pom.xml index d4ac3f5..6fb53d8 100644 --- a/pom.xml +++ b/pom.xml
@@ -21,14 +21,14 @@ <modelVersion>4.0.0</modelVersion> <parent> - <artifactId>flume-ng-sinks</artifactId> + <artifactId>flume-parent</artifactId> <groupId>org.apache.flume</groupId> <version>2.0.0-SNAPSHOT</version> </parent> - <groupId>org.apache.flume.flume-ng-sinks</groupId> - <artifactId>flume-http-sink</artifactId> - <name>Flume HTTP/S Sink</name> + <groupId>org.apache.flume</groupId> + <artifactId>flume-http</artifactId> + <name>Flume HTTP/S Sink and Source</name> <properties> <!-- TODO fix spotbugs/pmd violations --> @@ -112,7 +112,7 @@ <dependency> <groupId>org.mockito</groupId> - <artifactId>mockito-all</artifactId> + <artifactId>mockito-core</artifactId> <scope>test</scope> </dependency>
diff --git a/src/main/java/org/apache/flume/sink/http/HttpSink.java b/src/main/java/org/apache/flume/http/sink/HttpSink.java similarity index 99% rename from src/main/java/org/apache/flume/sink/http/HttpSink.java rename to src/main/java/org/apache/flume/http/sink/HttpSink.java index d4d7ad9..347fd34 100644 --- a/src/main/java/org/apache/flume/sink/http/HttpSink.java +++ b/src/main/java/org/apache/flume/http/sink/HttpSink.java
@@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.flume.sink.http; +package org.apache.flume.http.sink; import org.apache.flume.Channel; import org.apache.flume.Context;
diff --git a/src/main/java/org/apache/flume/sink/http/package-info.java b/src/main/java/org/apache/flume/http/sink/package-info.java similarity index 95% rename from src/main/java/org/apache/flume/sink/http/package-info.java rename to src/main/java/org/apache/flume/http/sink/package-info.java index 5b88f01..4c3bf3c 100644 --- a/src/main/java/org/apache/flume/sink/http/package-info.java +++ b/src/main/java/org/apache/flume/http/sink/package-info.java
@@ -20,4 +20,4 @@ * This package provides an HTTP sink for Flume so that events can be sent out * to a target HTTP endpoint. */ -package org.apache.flume.sink.http; +package org.apache.flume.http.sink;
diff --git a/src/main/java/org/apache/flume/http/source/BLOBHandler.java b/src/main/java/org/apache/flume/http/source/BLOBHandler.java new file mode 100644 index 0000000..9892dfe --- /dev/null +++ b/src/main/java/org/apache/flume/http/source/BLOBHandler.java
@@ -0,0 +1,101 @@ +/* + * 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.flume.http.source; + +import java.io.InputStream; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.commons.io.IOUtils; +import org.apache.commons.io.output.ByteArrayOutputStream; +import org.apache.flume.Context; +import org.apache.flume.Event; +import org.apache.flume.conf.LogPrivacyUtil; +import org.apache.flume.event.EventBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; + +import jakarta.servlet.http.HttpServletRequest; + +/** + * + * BLOBHandler for HTTPSource that accepts any binary stream of data as event. + * + */ +public class BLOBHandler implements HTTPSourceHandler { + + private static final Logger LOG = LoggerFactory.getLogger(BLOBHandler.class); + + private String commaSeparatedHeaders; + + private String[] mandatoryHeaders; + + public static final String MANDATORY_PARAMETERS = "mandatoryParameters"; + + public static final String DEFAULT_MANDATORY_PARAMETERS = ""; + + public static final String PARAMETER_SEPARATOR = ","; + + /** + * {@inheritDoc} + */ + @SuppressWarnings("unchecked") + @Override + public List<Event> getEvents(HttpServletRequest request) throws Exception { + Map<String, String> headers = new HashMap<String, String>(); + + InputStream inputStream = request.getInputStream(); + + Map<String, String[]> parameters = request.getParameterMap(); + for (String parameter : parameters.keySet()) { + String value = parameters.get(parameter)[0]; + if (LOG.isDebugEnabled() && LogPrivacyUtil.allowLogRawData()) { + LOG.debug("Setting Header [Key, Value] as [{},{}] ", parameter, value); + } + headers.put(parameter, value); + } + + for (String header : mandatoryHeaders) { + Preconditions.checkArgument( + headers.containsKey(header), "Please specify " + header + " parameter in the request."); + } + + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + try { + IOUtils.copy(inputStream, outputStream); + LOG.debug("Building an Event with stream of size -- {}", outputStream.size()); + Event event = EventBuilder.withBody(outputStream.toByteArray(), headers); + event.setHeaders(headers); + List<Event> eventList = new ArrayList<Event>(); + eventList.add(event); + return eventList; + } finally { + outputStream.close(); + inputStream.close(); + } + } + + @Override + public void configure(Context context) { + this.commaSeparatedHeaders = context.getString(MANDATORY_PARAMETERS, DEFAULT_MANDATORY_PARAMETERS); + this.mandatoryHeaders = commaSeparatedHeaders.split(PARAMETER_SEPARATOR); + } +}
diff --git a/src/main/java/org/apache/flume/http/source/HTTPBadRequestException.java b/src/main/java/org/apache/flume/http/source/HTTPBadRequestException.java new file mode 100644 index 0000000..12df71d --- /dev/null +++ b/src/main/java/org/apache/flume/http/source/HTTPBadRequestException.java
@@ -0,0 +1,42 @@ +/* + * 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.flume.http.source; + +import org.apache.flume.FlumeException; + +/** + * + * Exception thrown by an HTTP Handler if the request was not parsed correctly + * into an event because the request was not in the expected format. + * + */ +public class HTTPBadRequestException extends FlumeException { + + private static final long serialVersionUID = -3540764742069390951L; + + public HTTPBadRequestException(String msg) { + super(msg); + } + + public HTTPBadRequestException(String msg, Throwable th) { + super(msg, th); + } + + public HTTPBadRequestException(Throwable th) { + super(th); + } +}
diff --git a/src/main/java/org/apache/flume/http/source/HTTPSource.java b/src/main/java/org/apache/flume/http/source/HTTPSource.java new file mode 100644 index 0000000..b46e4db --- /dev/null +++ b/src/main/java/org/apache/flume/http/source/HTTPSource.java
@@ -0,0 +1,299 @@ +/* + * 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.flume.http.source; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.apache.flume.Context; +import org.apache.flume.Event; +import org.apache.flume.EventDrivenSource; +import org.apache.flume.conf.Configurable; +import org.apache.flume.exception.ChannelException; +import org.apache.flume.instrumentation.SourceCounter; +import org.apache.flume.source.SslContextAwareAbstractSource; +import org.apache.flume.tools.FlumeBeanConfigurator; +import org.apache.flume.tools.HTTPServerConstraintUtil; +import org.eclipse.jetty.http.HttpVersion; +import org.eclipse.jetty.jmx.MBeanContainer; +import org.eclipse.jetty.server.HttpConfiguration; +import org.eclipse.jetty.server.HttpConnectionFactory; +import org.eclipse.jetty.server.SecureRequestCustomizer; +import org.eclipse.jetty.server.Server; +import org.eclipse.jetty.server.ServerConnector; +import org.eclipse.jetty.server.SslConnectionFactory; +import org.eclipse.jetty.servlet.ServletContextHandler; +import org.eclipse.jetty.servlet.ServletHolder; +import org.eclipse.jetty.util.ssl.SslContextFactory; +import org.eclipse.jetty.util.thread.QueuedThreadPool; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; +import com.google.common.base.Throwables; + +import jakarta.servlet.http.HttpServlet; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; + +/** + * A source which accepts Flume Events by HTTP POST and GET. GET should be used + * for experimentation only. HTTP requests are converted into flume events by a + * pluggable "handler" which must implement the + * {@linkplain HTTPSourceHandler} interface. This handler takes a + * {@linkplain HttpServletRequest} and returns a list of flume events. + * + * The source accepts the following parameters: <p> <tt>port</tt>: port to which + * the server should bind. Mandatory <p> <tt>handler</tt>: the class that + * deserializes a HttpServletRequest into a list of flume events. This class + * must implement HTTPSourceHandler. Default: + * {@linkplain JSONHandler}. <p> <tt>handler.*</tt> Any configuration + * to be passed to the handler. <p> + * + * All events deserialized from one Http request are committed to the channel in + * one transaction, thus allowing for increased efficiency on channels like the + * file channel. If the handler throws an exception this source will return + * a HTTP status of 400. If the channel is full, or the source is unable to + * append events to the channel, the source will return a HTTP 503 - Temporarily + * unavailable status. + * + * A JSON handler which converts JSON objects to Flume events is provided. + * + */ +public class HTTPSource extends SslContextAwareAbstractSource implements EventDrivenSource, Configurable { + /* + * There are 2 ways of doing this: + * a. Have a static server instance and use connectors in each source + * which binds to the port defined for that source. + * b. Each source starts its own server instance, which binds to the source's + * port. + * + * b is more efficient than a because Jetty does not allow binding a + * servlet to a connector. So each request will need to go through each + * each of the handlers/servlet till the correct one is found. + * + */ + + private static final Logger LOG = LoggerFactory.getLogger(HTTPSource.class); + private volatile Integer port; + private volatile Server srv; + private volatile String host; + private HTTPSourceHandler handler; + private SourceCounter sourceCounter; + + private Context sourceContext; + + @Override + public void configure(Context context) { + configureSsl(context); + sourceContext = context; + try { + port = context.getInteger(HTTPSourceConfigurationConstants.CONFIG_PORT); + host = context.getString( + HTTPSourceConfigurationConstants.CONFIG_BIND, HTTPSourceConfigurationConstants.DEFAULT_BIND); + + Preconditions.checkState(host != null && !host.isEmpty(), "HTTPSource hostname specified is empty"); + Preconditions.checkNotNull(port, "HTTPSource requires a port number to be" + " specified"); + + String handlerClassName = context.getString( + HTTPSourceConfigurationConstants.CONFIG_HANDLER, + HTTPSourceConfigurationConstants.DEFAULT_HANDLER) + .trim(); + + @SuppressWarnings("unchecked") + Class<? extends HTTPSourceHandler> clazz = + (Class<? extends HTTPSourceHandler>) Class.forName(handlerClassName); + handler = clazz.getDeclaredConstructor().newInstance(); + + Map<String, String> subProps = + context.getSubProperties(HTTPSourceConfigurationConstants.CONFIG_HANDLER_PREFIX); + handler.configure(new Context(subProps)); + } catch (ClassNotFoundException ex) { + LOG.error("Error while configuring HTTPSource. Exception follows.", ex); + Throwables.propagate(ex); + } catch (ClassCastException ex) { + LOG.error("Deserializer is not an instance of HTTPSourceHandler." + + "Deserializer must implement HTTPSourceHandler."); + Throwables.propagate(ex); + } catch (Exception ex) { + LOG.error("Error configuring HTTPSource!", ex); + Throwables.propagate(ex); + } + if (sourceCounter == null) { + sourceCounter = new SourceCounter(getName()); + } + } + + @Override + public void start() { + Preconditions.checkState( + srv == null, + "Running HTTP Server found in source: " + getName() + + " before I started one." + + "Will not attempt to start."); + QueuedThreadPool threadPool = new QueuedThreadPool(); + if (sourceContext.getSubProperties("QueuedThreadPool.").size() > 0) { + FlumeBeanConfigurator.setConfigurationFields(threadPool, sourceContext); + } + srv = new Server(threadPool); + + // Register with JMX for advanced monitoring + MBeanContainer mbContainer = new MBeanContainer(ManagementFactory.getPlatformMBeanServer()); + srv.addEventListener(mbContainer); + srv.addBean(mbContainer); + + HttpConfiguration httpConfiguration = new HttpConfiguration(); + httpConfiguration.addCustomizer(new SecureRequestCustomizer()); + + FlumeBeanConfigurator.setConfigurationFields(httpConfiguration, sourceContext); + ServerConnector connector = getSslContextSupplier() + .get() + .map(sslContext -> { + SslContextFactory sslCtxFactory = new SslContextFactory(); + sslCtxFactory.setSslContext(sslContext); + sslCtxFactory.setExcludeProtocols(getExcludeProtocols().toArray(new String[] {})); + sslCtxFactory.setIncludeProtocols(getIncludeProtocols().toArray(new String[] {})); + sslCtxFactory.setExcludeCipherSuites( + getExcludeCipherSuites().toArray(new String[] {})); + sslCtxFactory.setIncludeCipherSuites( + getIncludeCipherSuites().toArray(new String[] {})); + + FlumeBeanConfigurator.setConfigurationFields(sslCtxFactory, sourceContext); + + httpConfiguration.setSecurePort(port); + httpConfiguration.setSecureScheme("https"); + + return new ServerConnector( + srv, + new SslConnectionFactory(sslCtxFactory, HttpVersion.HTTP_1_1.asString()), + new HttpConnectionFactory(httpConfiguration)); + }) + .orElse(new ServerConnector(srv, new HttpConnectionFactory(httpConfiguration))); + + connector.setPort(port); + connector.setHost(host); + connector.setReuseAddress(true); + + FlumeBeanConfigurator.setConfigurationFields(connector, sourceContext); + + srv.addConnector(connector); + + try { + ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS); + context.setContextPath("/"); + srv.setHandler(context); + + context.addServlet(new ServletHolder(new FlumeHTTPServlet()), "/"); + context.setSecurityHandler(HTTPServerConstraintUtil.enforceConstraints()); + srv.start(); + } catch (Exception ex) { + LOG.error("Error while starting HTTPSource. Exception follows.", ex); + Throwables.propagate(ex); + } + Preconditions.checkArgument(srv.isRunning()); + sourceCounter.start(); + super.start(); + } + + @Override + public void stop() { + try { + srv.stop(); + srv.join(); + srv = null; + } catch (Exception ex) { + LOG.error("Error while stopping HTTPSource. Exception follows.", ex); + } + sourceCounter.stop(); + LOG.info("Http source {} stopped. Metrics: {}", getName(), sourceCounter); + } + + private class FlumeHTTPServlet extends HttpServlet { + + private static final long serialVersionUID = 4891924863218790344L; + + @Override + public void doPost(HttpServletRequest request, HttpServletResponse response) throws IOException { + List<Event> events = Collections.emptyList(); // create empty list + try { + events = handler.getEvents(request); + } catch (HTTPBadRequestException ex) { + LOG.warn("Received bad request from client. ", ex); + sourceCounter.incrementEventReadFail(); + response.sendError(HttpServletResponse.SC_BAD_REQUEST, "Bad request from client."); + return; + } catch (Exception ex) { + LOG.warn("Deserializer threw unexpected exception. ", ex); + sourceCounter.incrementEventReadFail(); + response.sendError( + HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Deserializer threw unexpected exception."); + return; + } + sourceCounter.incrementAppendBatchReceivedCount(); + sourceCounter.addToEventReceivedCount(events.size()); + try { + getChannelProcessor().processEventBatch(events); + } catch (ChannelException ex) { + LOG.warn( + "Error appending event to channel. " + + "Channel might be full. Consider increasing the channel " + + "capacity or make sure the sinks perform faster.", + ex); + sourceCounter.incrementChannelWriteFail(); + response.sendError( + HttpServletResponse.SC_SERVICE_UNAVAILABLE, + "Error appending event to channel. Channel might be full."); + return; + } catch (Exception ex) { + LOG.warn("Unexpected error appending event to channel. ", ex); + sourceCounter.incrementGenericProcessingFail(); + response.sendError( + HttpServletResponse.SC_INTERNAL_SERVER_ERROR, + "Unexpected error while appending event to channel."); + return; + } + response.setCharacterEncoding(request.getCharacterEncoding()); + response.setStatus(HttpServletResponse.SC_OK); + response.flushBuffer(); + sourceCounter.incrementAppendBatchAcceptedCount(); + sourceCounter.addToEventAcceptedCount(events.size()); + } + + @Override + public void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException { + doPost(request, response); + } + } + + @Override + protected void configureSsl(Context context) { + handleDeprecatedParameter(context, "ssl", "enableSSL"); + handleDeprecatedParameter(context, "exclude-protocols", "excludeProtocols"); + handleDeprecatedParameter(context, "keystore-password", "keystorePassword"); + + super.configureSsl(context); + } + + private void handleDeprecatedParameter(Context context, String newParam, String oldParam) { + if (!context.containsKey(newParam) && context.containsKey(oldParam)) { + context.put(newParam, context.getString(oldParam)); + } + } +}
diff --git a/src/main/java/org/apache/flume/http/source/HTTPSourceConfigurationConstants.java b/src/main/java/org/apache/flume/http/source/HTTPSourceConfigurationConstants.java new file mode 100644 index 0000000..d8b381e --- /dev/null +++ b/src/main/java/org/apache/flume/http/source/HTTPSourceConfigurationConstants.java
@@ -0,0 +1,45 @@ +/* + * 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.flume.http.source; + +/** + * + */ +public class HTTPSourceConfigurationConstants { + + public static final String CONFIG_PORT = "port"; + public static final String CONFIG_HANDLER = "handler"; + public static final String CONFIG_HANDLER_PREFIX = CONFIG_HANDLER + "."; + + public static final String CONFIG_BIND = "bind"; + + public static final String DEFAULT_BIND = "0.0.0.0"; + + public static final String DEFAULT_HANDLER = "org.apache.flume.source.http.JSONHandler"; + + @Deprecated + public static final String SSL_KEYSTORE = "keystore"; + + @Deprecated + public static final String SSL_KEYSTORE_PASSWORD = "keystorePassword"; + + @Deprecated + public static final String SSL_ENABLED = "enableSSL"; + + @Deprecated + public static final String EXCLUDE_PROTOCOLS = "excludeProtocols"; +}
diff --git a/src/main/java/org/apache/flume/http/source/HTTPSourceHandler.java b/src/main/java/org/apache/flume/http/source/HTTPSourceHandler.java new file mode 100644 index 0000000..c8bbada --- /dev/null +++ b/src/main/java/org/apache/flume/http/source/HTTPSourceHandler.java
@@ -0,0 +1,44 @@ +/* + * 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.flume.http.source; + +import java.util.List; + +import org.apache.flume.Event; +import org.apache.flume.conf.Configurable; + +import jakarta.servlet.http.HttpServletRequest; + +/** + * + */ +public interface HTTPSourceHandler extends Configurable { + + /** + * Takes an {@linkplain HttpServletRequest} and returns a list of Flume + * Events. If this request cannot be parsed into Flume events based on the + * format this method will throw an exception. This method may also throw an + * exception if there is some sort of other error. <p> + * + * @param request The request to be parsed into Flume events. + * @return List of Flume events generated from the request. + * @throws HTTPBadRequestException If the was not parsed correctly into an + * event because the request was not in the expected format. + * @throws Exception If there was an unexpected error. + */ + public List<Event> getEvents(HttpServletRequest request) throws HTTPBadRequestException, Exception; +}
diff --git a/src/main/java/org/apache/flume/http/source/JSONHandler.java b/src/main/java/org/apache/flume/http/source/JSONHandler.java new file mode 100644 index 0000000..96479dc --- /dev/null +++ b/src/main/java/org/apache/flume/http/source/JSONHandler.java
@@ -0,0 +1,136 @@ +/* + * 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.flume.http.source; + +import java.io.BufferedReader; +import java.lang.reflect.Type; +import java.nio.charset.UnsupportedCharsetException; +import java.util.ArrayList; +import java.util.List; + +import org.apache.flume.Context; +import org.apache.flume.Event; +import org.apache.flume.event.EventBuilder; +import org.apache.flume.event.JSONEvent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.JsonSyntaxException; +import com.google.gson.reflect.TypeToken; + +import jakarta.servlet.http.HttpServletRequest; + +/** + * JSONHandler for HTTPSource that accepts an array of events. + * + * This handler throws exception if the deserialization fails because of bad + * format or any other reason. + * + * Each event must be encoded as a map with two key-value pairs. <p> 1. headers + * - the key for this key-value pair is "headers". The value for this key is + * another map, which represent the event headers. These headers are inserted + * into the Flume event as is. <p> 2. body - The body is a string which + * represents the body of the event. The key for this key-value pair is "body". + * All key-value pairs are considered to be headers. An example: <p> [{"headers" + * : {"a":"b", "c":"d"},"body": "random_body"}, {"headers" : {"e": "f"},"body": + * "random_body2"}] <p> would be interpreted as the following two flume events: + * <p> * Event with body: "random_body" (in UTF-8/UTF-16/UTF-32 encoded bytes) + * and headers : (a:b, c:d) <p> * + * Event with body: "random_body2" (in UTF-8/UTF-16/UTF-32 encoded bytes) and + * headers : (e:f) <p> + * + * The charset of the body is read from the request and used. If no charset is + * set in the request, then the charset is assumed to be JSON's default - UTF-8. + * The JSON handler supports UTF-8, UTF-16 and UTF-32. + * + * To set the charset, the request must have content type specified as + * "application/json; charset=UTF-8" (replace UTF-8 with UTF-16 or UTF-32 as + * required). + * + * One way to create an event in the format expected by this handler, is to + * use {@linkplain JSONEvent} and use {@linkplain Gson} to create the JSON + * string using the + * {@linkplain Gson#toJson(Object, Type) } + * method. The type token to pass as the 2nd argument of this method + * for list of events can be created by: <p> + * {@code + * Type type = new TypeToken<List<JSONEvent>>() {}.getType(); + * } + */ +public class JSONHandler implements HTTPSourceHandler { + + private static final Logger LOG = LoggerFactory.getLogger(JSONHandler.class); + private final Type listType = new TypeToken<List<JSONEvent>>() {}.getType(); + private final Gson gson; + + public JSONHandler() { + gson = new GsonBuilder().disableHtmlEscaping().create(); + } + + /** + * {@inheritDoc} + */ + @Override + public List<Event> getEvents(HttpServletRequest request) throws Exception { + BufferedReader reader = request.getReader(); + String charset = request.getCharacterEncoding(); + // UTF-8 is default for JSON. If no charset is specified, UTF-8 is to + // be assumed. + if (charset == null) { + LOG.debug("Charset is null, default charset of UTF-8 will be used."); + charset = "UTF-8"; + } else if (!(charset.equalsIgnoreCase("utf-8") + || charset.equalsIgnoreCase("utf-16") + || charset.equalsIgnoreCase("utf-32"))) { + LOG.error( + "Unsupported character set in request {}. " + + "JSON handler supports UTF-8, " + + "UTF-16 and UTF-32 only.", + charset); + throw new UnsupportedCharsetException("JSON handler supports UTF-8, " + "UTF-16 and UTF-32 only."); + } + + /* + * Gson throws Exception if the data is not parseable to JSON. + * Need not catch it since the source will catch it and return error. + */ + List<Event> eventList = new ArrayList<Event>(0); + try { + eventList = gson.fromJson(reader, listType); + } catch (JsonSyntaxException ex) { + throw new HTTPBadRequestException("Request has invalid JSON Syntax.", ex); + } + + for (Event e : eventList) { + ((JSONEvent) e).setCharset(charset); + } + return getSimpleEvents(eventList); + } + + @Override + public void configure(Context context) {} + + private List<Event> getSimpleEvents(List<Event> events) { + List<Event> newEvents = new ArrayList<Event>(events.size()); + for (Event e : events) { + newEvents.add(EventBuilder.withBody(e.getBody(), e.getHeaders())); + } + return newEvents; + } +}
diff --git a/src/test/java/org/apache/flume/sink/http/TestHttpSink.java b/src/test/java/org/apache/flume/http/sink/TestHttpSink.java similarity index 99% rename from src/test/java/org/apache/flume/sink/http/TestHttpSink.java rename to src/test/java/org/apache/flume/http/sink/TestHttpSink.java index d8fcde2..eee80b1 100644 --- a/src/test/java/org/apache/flume/sink/http/TestHttpSink.java +++ b/src/test/java/org/apache/flume/http/sink/TestHttpSink.java
@@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.flume.sink.http; +package org.apache.flume.http.sink; import org.apache.flume.Channel; import org.apache.flume.Context;
diff --git a/src/test/java/org/apache/flume/sink/http/TestHttpSinkIT.java b/src/test/java/org/apache/flume/http/sink/TestHttpSinkIT.java similarity index 99% rename from src/test/java/org/apache/flume/sink/http/TestHttpSinkIT.java rename to src/test/java/org/apache/flume/http/sink/TestHttpSinkIT.java index 9e2c356..57125e2 100644 --- a/src/test/java/org/apache/flume/sink/http/TestHttpSinkIT.java +++ b/src/test/java/org/apache/flume/http/sink/TestHttpSinkIT.java
@@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.flume.sink.http; +package org.apache.flume.http.sink; import com.github.tomakehurst.wiremock.global.RequestDelaySpec; import com.github.tomakehurst.wiremock.http.Fault;
diff --git a/src/test/java/org/apache/flume/http/source/FlumeHttpServletRequestWrapper.java b/src/test/java/org/apache/flume/http/source/FlumeHttpServletRequestWrapper.java new file mode 100644 index 0000000..6cf7a5a --- /dev/null +++ b/src/test/java/org/apache/flume/http/source/FlumeHttpServletRequestWrapper.java
@@ -0,0 +1,407 @@ +/* + * 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.flume.http.source; + +import java.io.BufferedReader; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.UnsupportedEncodingException; +import java.security.Principal; +import java.util.Collection; +import java.util.Enumeration; +import java.util.Locale; +import java.util.Map; + +import jakarta.servlet.AsyncContext; +import jakarta.servlet.DispatcherType; +import jakarta.servlet.RequestDispatcher; +import jakarta.servlet.ServletContext; +import jakarta.servlet.ServletException; +import jakarta.servlet.ServletInputStream; +import jakarta.servlet.ServletRequest; +import jakarta.servlet.ServletResponse; +import jakarta.servlet.http.Cookie; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import jakarta.servlet.http.HttpSession; +import jakarta.servlet.http.HttpUpgradeHandler; +import jakarta.servlet.http.Part; + +/** + * + */ +class FlumeHttpServletRequestWrapper implements HttpServletRequest { + + private BufferedReader reader; + + String charset; + + public FlumeHttpServletRequestWrapper(String data, String charset) throws UnsupportedEncodingException { + reader = new BufferedReader(new InputStreamReader(new ByteArrayInputStream(data.getBytes(charset)), charset)); + this.charset = charset; + } + + public FlumeHttpServletRequestWrapper(String data) throws UnsupportedEncodingException { + this(data, "UTF-8"); + } + + @Override + public String getAuthType() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Cookie[] getCookies() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public long getDateHeader(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getHeader(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Enumeration<String> getHeaders(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Enumeration<String> getHeaderNames() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public int getIntHeader(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getMethod() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getPathInfo() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getPathTranslated() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getContextPath() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getQueryString() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getRemoteUser() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isUserInRole(String role) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Principal getUserPrincipal() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getRequestedSessionId() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getRequestURI() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public StringBuffer getRequestURL() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getServletPath() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public HttpSession getSession(boolean create) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public HttpSession getSession() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isRequestedSessionIdValid() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isRequestedSessionIdFromCookie() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isRequestedSessionIdFromURL() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isRequestedSessionIdFromUrl() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Object getAttribute(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Enumeration<String> getAttributeNames() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getCharacterEncoding() { + return charset; + } + + @Override + public void setCharacterEncoding(String env) throws UnsupportedEncodingException { + this.charset = env; + } + + @Override + public int getContentLength() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getContentType() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public ServletInputStream getInputStream() throws IOException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getParameter(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Enumeration<String> getParameterNames() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String[] getParameterValues(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Map<String, String[]> getParameterMap() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getProtocol() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getScheme() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getServerName() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public int getServerPort() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public BufferedReader getReader() throws IOException { + return reader; + } + + @Override + public String getRemoteAddr() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getRemoteHost() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public void setAttribute(String name, Object o) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public void removeAttribute(String name) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Locale getLocale() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Enumeration<Locale> getLocales() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isSecure() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public RequestDispatcher getRequestDispatcher(String path) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getRealPath(String path) { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public int getRemotePort() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getLocalName() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String getLocalAddr() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public int getLocalPort() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public AsyncContext getAsyncContext() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public long getContentLengthLong() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public DispatcherType getDispatcherType() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public ServletContext getServletContext() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isAsyncStarted() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean isAsyncSupported() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public AsyncContext startAsync() throws IllegalStateException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public AsyncContext startAsync(ServletRequest arg0, ServletResponse arg1) throws IllegalStateException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public boolean authenticate(HttpServletResponse arg0) throws IOException, ServletException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public String changeSessionId() { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Part getPart(String arg0) throws IOException, ServletException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public Collection<Part> getParts() throws IOException, ServletException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public void login(String arg0, String arg1) throws ServletException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public void logout() throws ServletException { + throw new UnsupportedOperationException("Not supported yet."); + } + + @Override + public <T extends HttpUpgradeHandler> T upgrade(Class<T> arg0) throws IOException, ServletException { + throw new UnsupportedOperationException("Not supported yet."); + } +}
diff --git a/src/test/java/org/apache/flume/http/source/TestBLOBHandler.java b/src/test/java/org/apache/flume/http/source/TestBLOBHandler.java new file mode 100644 index 0000000..507a63b --- /dev/null +++ b/src/test/java/org/apache/flume/http/source/TestBLOBHandler.java
@@ -0,0 +1,174 @@ +/* + * 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.flume.http.source; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.flume.Context; +import org.apache.flume.Event; +import org.junit.Before; +import org.junit.Test; + +import jakarta.servlet.ReadListener; +import jakarta.servlet.ServletInputStream; +import jakarta.servlet.http.HttpServletRequest; +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * + */ +public class TestBLOBHandler { + + HTTPSourceHandler handler; + + @Before + public void setUp() { + handler = new BLOBHandler(); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void testCSVData() throws Exception { + Map requestParameterMap = new HashMap(); + requestParameterMap.put("param1", new String[] {"value1"}); + requestParameterMap.put("param2", new String[] {"value2"}); + + HttpServletRequest req = mock(HttpServletRequest.class); + final String csvData = "a,b,c"; + + ServletInputStream servletInputStream = + new DelegatingServletInputStream(new ByteArrayInputStream(csvData.getBytes())); + + when(req.getInputStream()).thenReturn(servletInputStream); + when(req.getParameterMap()).thenReturn(requestParameterMap); + + Context context = mock(Context.class); + when(context.getString(BLOBHandler.MANDATORY_PARAMETERS, BLOBHandler.DEFAULT_MANDATORY_PARAMETERS)) + .thenReturn("param1,param2"); + + handler.configure(context); + List<Event> deserialized = handler.getEvents(req); + assertEquals(1, deserialized.size()); + Event e = deserialized.get(0); + + assertEquals(new String(e.getBody()), csvData); + assertEquals(e.getHeaders().get("param1"), "value1"); + assertEquals(e.getHeaders().get("param2"), "value2"); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + public void testTabData() throws Exception { + Map requestParameterMap = new HashMap(); + requestParameterMap.put("param1", new String[] {"value1"}); + + HttpServletRequest req = mock(HttpServletRequest.class); + final String tabData = "a\tb\tc"; + + ServletInputStream servletInputStream = + new DelegatingServletInputStream(new ByteArrayInputStream(tabData.getBytes())); + + when(req.getInputStream()).thenReturn(servletInputStream); + when(req.getParameterMap()).thenReturn(requestParameterMap); + + Context context = mock(Context.class); + when(context.getString(BLOBHandler.MANDATORY_PARAMETERS, BLOBHandler.DEFAULT_MANDATORY_PARAMETERS)) + .thenReturn("param1"); + + handler.configure(context); + + List<Event> deserialized = handler.getEvents(req); + assertEquals(1, deserialized.size()); + Event e = deserialized.get(0); + + assertEquals(new String(e.getBody()), tabData); + assertEquals(e.getHeaders().get("param1"), "value1"); + } + + @SuppressWarnings({"rawtypes"}) + @Test(expected = IllegalArgumentException.class) + public void testMissingParameters() throws Exception { + Map requestParameterMap = new HashMap(); + + HttpServletRequest req = mock(HttpServletRequest.class); + final String tabData = "a\tb\tc"; + + ServletInputStream servletInputStream = + new DelegatingServletInputStream(new ByteArrayInputStream(tabData.getBytes())); + + when(req.getInputStream()).thenReturn(servletInputStream); + when(req.getParameterMap()).thenReturn(requestParameterMap); + + Context context = mock(Context.class); + when(context.getString(BLOBHandler.MANDATORY_PARAMETERS, BLOBHandler.DEFAULT_MANDATORY_PARAMETERS)) + .thenReturn("param1"); + + handler.configure(context); + + handler.getEvents(req); + } + + class DelegatingServletInputStream extends ServletInputStream { + + private final InputStream sourceStream; + + /** + * Create a DelegatingServletInputStream for the given source stream. + * + * @param sourceStream + * the source stream (never <code>null</code>) + */ + public DelegatingServletInputStream(InputStream sourceStream) { + this.sourceStream = sourceStream; + } + + /** + * Return the underlying source stream (never <code>null</code>). + */ + public final InputStream getSourceStream() { + return this.sourceStream; + } + + public int read() throws IOException { + return this.sourceStream.read(); + } + + public void close() throws IOException { + super.close(); + this.sourceStream.close(); + } + + public boolean isFinished() { + throw new UnsupportedOperationException("Not supported yet."); + } + + public boolean isReady() { + throw new UnsupportedOperationException("Not supported yet."); + } + + public void setReadListener(ReadListener arg0) { + throw new UnsupportedOperationException("Not supported yet."); + } + } +}
diff --git a/src/test/java/org/apache/flume/http/source/TestDefaultSourceFactory.java b/src/test/java/org/apache/flume/http/source/TestDefaultSourceFactory.java new file mode 100644 index 0000000..f1b0ce7 --- /dev/null +++ b/src/test/java/org/apache/flume/http/source/TestDefaultSourceFactory.java
@@ -0,0 +1,45 @@ +/* + * 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.flume.http.source; + +import org.apache.flume.Source; +import org.apache.flume.SourceFactory; +import org.apache.flume.source.DefaultSourceFactory; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class TestDefaultSourceFactory { + + private SourceFactory sourceFactory; + + @Before + public void setUp() { + sourceFactory = new DefaultSourceFactory(); + } + + private void verifySourceCreation(String name, String type, Class<?> typeClass) throws Exception { + Source src = sourceFactory.create(name, type); + Assert.assertNotNull(src); + Assert.assertTrue(typeClass.isInstance(src)); + } + + @Test + public void testSourceCreation() throws Exception { + verifySourceCreation("http-src", "http", HTTPSource.class); + } +}
diff --git a/src/test/java/org/apache/flume/http/source/TestHTTPSource.java b/src/test/java/org/apache/flume/http/source/TestHTTPSource.java new file mode 100644 index 0000000..1471e1d --- /dev/null +++ b/src/test/java/org/apache/flume/http/source/TestHTTPSource.java
@@ -0,0 +1,734 @@ +/* + * 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.flume.http.source; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.reflect.Type; +import java.net.HttpURLConnection; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.URL; +import java.net.UnknownHostException; +import java.security.SecureRandom; +import java.security.cert.CertificateException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import javax.management.MBeanServer; +import javax.management.MalformedObjectNameException; +import javax.management.ObjectInstance; +import javax.management.ObjectName; +import javax.management.Query; +import javax.management.QueryExp; +import javax.net.ssl.HostnameVerifier; +import javax.net.ssl.HttpsURLConnection; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSession; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; + +import org.apache.flume.Channel; +import org.apache.flume.ChannelSelector; +import org.apache.flume.Context; +import org.apache.flume.Event; +import org.apache.flume.Transaction; +import org.apache.flume.channel.ChannelProcessor; +import org.apache.flume.channel.MemoryChannel; +import org.apache.flume.channel.ReplicatingChannelSelector; +import org.apache.flume.conf.Configurables; +import org.apache.flume.event.JSONEvent; +import org.apache.flume.instrumentation.SourceCounter; +import org.apache.flume.util.Whitebox; +import org.apache.http.HttpResponse; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpOptions; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpRequestBase; +import org.apache.http.client.methods.HttpTrace; +import org.apache.http.conn.ssl.NoopHostnameVerifier; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.HttpClientBuilder; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.Mockito; + +import com.google.common.collect.Maps; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; + +import jakarta.servlet.http.HttpServletResponse; +import static org.fest.reflect.core.Reflection.field; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doThrow; + +/** + * + */ +public class TestHTTPSource { + + private static HTTPSource httpSource; + private static HTTPSource httpsSource; + private static HTTPSource httpsGlobalKeystoreSource; + + private static Channel httpChannel; + private static Channel httpsChannel; + private static Channel httpsGlobalKeystoreChannel; + private static int httpPort; + private static int httpsPort; + private static int httpsGlobalKeystorePort; + private HttpClient httpClient; + private HttpPost postRequest; + + private static int findFreePort() throws IOException { + ServerSocket socket = new ServerSocket(0); + int port = socket.getLocalPort(); + socket.close(); + return port; + } + + private static Context getDefaultNonSecureContext(int port) throws IOException { + Context ctx = new Context(); + ctx.put(HTTPSourceConfigurationConstants.CONFIG_BIND, "0.0.0.0"); + ctx.put(HTTPSourceConfigurationConstants.CONFIG_PORT, String.valueOf(port)); + ctx.put("QueuedThreadPool.MaxThreads", "100"); + return ctx; + } + + private static Context getDefaultSecureContext(int port) throws IOException { + Context sslContext = new Context(); + sslContext.put(HTTPSourceConfigurationConstants.CONFIG_PORT, String.valueOf(port)); + sslContext.put(HTTPSourceConfigurationConstants.SSL_ENABLED, "true"); + sslContext.put(HTTPSourceConfigurationConstants.SSL_KEYSTORE_PASSWORD, "password"); + sslContext.put(HTTPSourceConfigurationConstants.SSL_KEYSTORE, "src/test/resources/jettykeystore"); + return sslContext; + } + + private static Context getDefaultSecureContextGlobalKeystore(int port) throws IOException { + System.setProperty("javax.net.ssl.keyStore", "src/test/resources/jettykeystore"); + System.setProperty("javax.net.ssl.keyStorePassword", "password"); + + Context sslContext = new Context(); + sslContext.put(HTTPSourceConfigurationConstants.CONFIG_PORT, String.valueOf(port)); + sslContext.put(HTTPSourceConfigurationConstants.SSL_ENABLED, "true"); + return sslContext; + } + + @BeforeClass + public static void setUpClass() throws Exception { + httpSource = new HTTPSource(); + httpChannel = new MemoryChannel(); + httpPort = findFreePort(); + configureSourceAndChannel(httpSource, httpChannel, getDefaultNonSecureContext(httpPort)); + httpChannel.start(); + httpSource.start(); + + httpsSource = new HTTPSource(); + httpsChannel = new MemoryChannel(); + httpsPort = findFreePort(); + configureSourceAndChannel(httpsSource, httpsChannel, getDefaultSecureContext(httpsPort)); + httpsChannel.start(); + httpsSource.start(); + + httpsGlobalKeystoreSource = new HTTPSource(); + httpsGlobalKeystoreChannel = new MemoryChannel(); + httpsGlobalKeystorePort = findFreePort(); + configureSourceAndChannel( + httpsGlobalKeystoreSource, + httpsGlobalKeystoreChannel, + getDefaultSecureContextGlobalKeystore(httpsGlobalKeystorePort)); + httpsGlobalKeystoreChannel.start(); + httpsGlobalKeystoreSource.start(); + + System.clearProperty("javax.net.ssl.keyStore"); + System.clearProperty("javax.net.ssl.keyStorePassword"); + } + + private static void configureSourceAndChannel(HTTPSource source, Channel channel, Context context) { + Context channelContext = new Context(); + channelContext.put("capacity", "100"); + Configurables.configure(channel, channelContext); + Configurables.configure(source, context); + + ChannelSelector rcs1 = new ReplicatingChannelSelector(); + rcs1.setChannels(Collections.singletonList(channel)); + + source.setChannelProcessor(new ChannelProcessor(rcs1)); + } + + @AfterClass + public static void tearDownClass() throws Exception { + httpSource.stop(); + httpChannel.stop(); + httpsSource.stop(); + httpsChannel.stop(); + httpsGlobalKeystoreSource.stop(); + httpsGlobalKeystoreChannel.stop(); + } + + @Before + public void setUp() { + HttpClientBuilder builder = HttpClientBuilder.create(); + httpClient = builder.build(); + postRequest = new HttpPost("http://0.0.0.0:" + httpPort); + SourceCounter sc = (SourceCounter) Whitebox.getInternalState(httpSource, "sourceCounter"); + sc.start(); + } + + @After + public void tearDown() { + SourceCounter sc = (SourceCounter) Whitebox.getInternalState(httpSource, "sourceCounter"); + sc.stop(); + } + + @Test + public void testSimple() throws IOException, InterruptedException { + + StringEntity input = new StringEntity("[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"); + // if we do not set the content type to JSON, the client will use + // ISO-8859-1 as the charset. JSON standard does not support this. + input.setContentType("application/json"); + postRequest.setEntity(input); + + HttpResponse response = httpClient.execute(postRequest); + + Assert.assertEquals(HttpServletResponse.SC_OK, response.getStatusLine().getStatusCode()); + Transaction tx = httpChannel.getTransaction(); + tx.begin(); + Event e = httpChannel.take(); + Assert.assertNotNull(e); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + + e = httpChannel.take(); + Assert.assertNotNull(e); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-8")); + tx.commit(); + tx.close(); + } + + @Test + public void testTrace() throws Exception { + doTestForbidden(new HttpTrace("http://0.0.0.0:" + httpPort)); + } + + @Test + public void testOptions() throws Exception { + doTestForbidden(new HttpOptions("http://0.0.0.0:" + httpPort)); + } + + private void doTestForbidden(HttpRequestBase request) throws Exception { + HttpResponse response = httpClient.execute(request); + Assert.assertEquals( + HttpServletResponse.SC_FORBIDDEN, response.getStatusLine().getStatusCode()); + } + + @Test + public void testSimpleUTF16() throws IOException, InterruptedException { + + StringEntity input = new StringEntity( + "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]", + "UTF-16"); + input.setContentType("application/json; charset=utf-16"); + postRequest.setEntity(input); + + HttpResponse response = httpClient.execute(postRequest); + + Assert.assertEquals(HttpServletResponse.SC_OK, response.getStatusLine().getStatusCode()); + Transaction tx = httpChannel.getTransaction(); + tx.begin(); + Event e = httpChannel.take(); + Assert.assertNotNull(e); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-16")); + + e = httpChannel.take(); + Assert.assertNotNull(e); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-16")); + tx.commit(); + tx.close(); + } + + @Test + public void testInvalid() throws Exception { + StringEntity input = new StringEntity("[{\"a\": \"b\",[\"d\":\"e\"],\"body\": \"random_body\"}," + + "{\"e\": \"f\",\"body\": \"random_body2\"}]"); + input.setContentType("application/json"); + postRequest.setEntity(input); + HttpResponse response = httpClient.execute(postRequest); + + Assert.assertEquals( + HttpServletResponse.SC_BAD_REQUEST, response.getStatusLine().getStatusCode()); + SourceCounter sc = (SourceCounter) Whitebox.getInternalState(httpSource, "sourceCounter"); + Assert.assertEquals(1, sc.getEventReadFail()); + } + + @Test + public void testBigBatchDeserializationUTF8() throws Exception { + testBatchWithVariousEncoding("UTF-8"); + } + + @Test + public void testBigBatchDeserializationUTF16() throws Exception { + testBatchWithVariousEncoding("UTF-16"); + } + + @Test + public void testBigBatchDeserializationUTF32() throws Exception { + testBatchWithVariousEncoding("UTF-32"); + } + + @Test + public void testCounterGenericFail() throws Exception { + ChannelProcessor cp = Mockito.mock(ChannelProcessor.class); + doThrow(new RuntimeException("dummy")).when(cp).processEventBatch(anyList()); + ChannelProcessor oldCp = httpSource.getChannelProcessor(); + httpSource.setChannelProcessor(cp); + testBatchWithVariousEncoding("UTF-8"); + SourceCounter sc = (SourceCounter) Whitebox.getInternalState(httpSource, "sourceCounter"); + Assert.assertEquals(1, sc.getGenericProcessingFail()); + httpSource.setChannelProcessor(oldCp); + } + + @Test + public void testSingleEvent() throws Exception { + StringEntity input = new StringEntity("[{\"headers\" : {\"a\": \"b\"},\"body\":" + " \"random_body\"}]"); + input.setContentType("application/json"); + postRequest.setEntity(input); + + httpClient.execute(postRequest); + Transaction tx = httpChannel.getTransaction(); + tx.begin(); + Event e = httpChannel.take(); + Assert.assertNotNull(e); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + tx.commit(); + tx.close(); + } + + /** + * First test that the unconfigured behaviour is as-expected, then add configurations + * to a new channel and observe the difference. + * For some of the properties, the most convenient way to test is using the MBean interface + * We test all of HttpConfiguration, ServerConnector, QueuedThreadPool and SslContextFactory + * sub-configurations (but not all properties) + */ + @Test + public void testConfigurables() throws Exception { + StringEntity input = new StringEntity("[{\"headers\" : {\"a\": \"b\"},\"body\":" + " \"random_body\"}]"); + input.setContentType("application/json"); + postRequest.setEntity(input); + + HttpResponse resp = httpClient.execute(postRequest); + + // Testing default behaviour (to not provided X-Powered-By, but to provide Server headers) + Assert.assertTrue(resp.getHeaders("X-Powered-By").length == 0); + Assert.assertTrue(resp.getHeaders("Server").length == 1); + + Transaction tx = httpChannel.getTransaction(); + tx.begin(); + Event e = httpChannel.take(); + Assert.assertNotNull(e); + tx.commit(); + tx.close(); + Assert.assertTrue(findMBeans("org.eclipse.jetty.util.thread:type=queuedthreadpool,*", "maxThreads", 123) + .size() + == 0); + Assert.assertTrue(findMBeans("org.eclipse.jetty.server:type=serverconnector,*", "acceptQueueSize", 22) + .size() + == 0); + + int newPort = findFreePort(); + Context configuredSourceContext = getDefaultNonSecureContext(newPort); + configuredSourceContext.put("HttpConfiguration.sendServerVersion", "false"); + configuredSourceContext.put("HttpConfiguration.sendXPoweredBy", "true"); + configuredSourceContext.put("ServerConnector.acceptQueueSize", "22"); + configuredSourceContext.put("QueuedThreadPool.maxThreads", "123"); + + HTTPSource newSource = new HTTPSource(); + Channel newChannel = new MemoryChannel(); + configureSourceAndChannel(newSource, newChannel, configuredSourceContext); + newChannel.start(); + newSource.start(); + + HttpPost newPostRequest = new HttpPost("http://0.0.0.0:" + newPort); + + resp = httpClient.execute(newPostRequest); + Assert.assertTrue(resp.getHeaders("X-Powered-By").length > 0); + Assert.assertTrue(resp.getHeaders("Server").length == 0); + Assert.assertTrue(findMBeans("org.eclipse.jetty.util.thread:type=queuedthreadpool,*", "maxThreads", 123) + .size() + == 1); + Assert.assertTrue(findMBeans("org.eclipse.jetty.server:type=serverconnector,*", "acceptQueueSize", 22) + .size() + == 1); + + newSource.stop(); + newChannel.stop(); + + // Configure SslContextFactory with junk protocols (expect failure) + newPort = findFreePort(); + configuredSourceContext = getDefaultSecureContext(newPort); + configuredSourceContext.put("SslContextFactory.IncludeProtocols", "abc def"); + + newSource = new HTTPSource(); + newChannel = new MemoryChannel(); + + configureSourceAndChannel(newSource, newChannel, configuredSourceContext); + + newChannel.start(); + newSource.start(); + + newPostRequest = new HttpPost("http://0.0.0.0:" + newPort); + try { + doTestHttps(null, newPort, httpsChannel); + // We are testing that this fails because we've deliberately configured the wrong protocols + Assert.assertTrue(false); + } catch (AssertionError ex) { + // no-op + } + newSource.stop(); + newChannel.stop(); + } + + @Test + public void testFullChannel() throws Exception { + HttpResponse response = putWithEncoding("UTF-8", 150).response; + Assert.assertEquals( + HttpServletResponse.SC_SERVICE_UNAVAILABLE, + response.getStatusLine().getStatusCode()); + SourceCounter sc = (SourceCounter) Whitebox.getInternalState(httpSource, "sourceCounter"); + Assert.assertEquals(1, sc.getChannelWriteFail()); + } + + @Test + public void testFail() throws Exception { + HTTPSourceHandler handler = + field("handler").ofType(HTTPSourceHandler.class).in(httpSource).get(); + // Cause an exception in the source - this is equivalent to any exception + // thrown by the handler since the handler is called inside a try-catch + field("handler").ofType(HTTPSourceHandler.class).in(httpSource).set(null); + HttpResponse response = putWithEncoding("UTF-8", 1).response; + Assert.assertEquals( + HttpServletResponse.SC_INTERNAL_SERVER_ERROR, + response.getStatusLine().getStatusCode()); + // Set the original handler back so tests don't fail after this runs. + field("handler").ofType(HTTPSourceHandler.class).in(httpSource).set(handler); + } + + @Test + public void testMBeans() throws Exception { + MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer(); + ObjectName objectName = new ObjectName("org.eclipse.jetty.*:*"); + Set<ObjectInstance> queryMBeans = mbeanServer.queryMBeans(objectName, null); + Assert.assertTrue(queryMBeans.size() > 0); + } + + @Test + public void testHandlerThrowingException() throws Exception { + // This will cause the handler to throw an + // UnsupportedCharsetException. + HttpResponse response = putWithEncoding("ISO-8859-1", 150).response; + Assert.assertEquals( + HttpServletResponse.SC_INTERNAL_SERVER_ERROR, + response.getStatusLine().getStatusCode()); + } + + private Set<ObjectInstance> findMBeans(String name, String attribute, int value) + throws MalformedObjectNameException { + MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer(); + ObjectName objectName = new ObjectName(name); + QueryExp q = Query.eq(Query.attr(attribute), Query.value(value)); + return mbeanServer.queryMBeans(objectName, q); + } + + private ResultWrapper putWithEncoding(String encoding, int n) throws Exception { + Type listType = new TypeToken<List<JSONEvent>>() {}.getType(); + List<JSONEvent> events = new ArrayList<JSONEvent>(); + Random rand = new Random(); + for (int i = 0; i < n; i++) { + Map<String, String> input = Maps.newHashMap(); + for (int j = 0; j < 10; j++) { + input.put(String.valueOf(i) + String.valueOf(j), String.valueOf(i)); + } + JSONEvent e = new JSONEvent(); + e.setHeaders(input); + e.setBody(String.valueOf(rand.nextGaussian()).getBytes(encoding)); + events.add(e); + } + Gson gson = new Gson(); + String json = gson.toJson(events, listType); + StringEntity input = new StringEntity(json); + input.setContentType("application/json; charset=" + encoding); + postRequest.setEntity(input); + HttpResponse resp = httpClient.execute(postRequest); + return new ResultWrapper(resp, events); + } + + @Test + public void testHttps() throws Exception { + doTestHttps(null, httpsPort, httpsChannel); + } + + @Test(expected = javax.net.ssl.SSLHandshakeException.class) + public void testHttpsSSLv3() throws Exception { + doTestHttps("SSLv3", httpsPort, httpsChannel); + } + + @Test + public void testHttpsGlobalKeystore() throws Exception { + doTestHttps(null, httpsGlobalKeystorePort, httpsGlobalKeystoreChannel); + } + + private void doTestHttps(String protocol, int port, Channel channel) throws Exception { + Type listType = new TypeToken<List<JSONEvent>>() {}.getType(); + List<JSONEvent> events = new ArrayList<JSONEvent>(); + Random rand = new Random(); + for (int i = 0; i < 10; i++) { + Map<String, String> input = Maps.newHashMap(); + for (int j = 0; j < 10; j++) { + input.put(String.valueOf(i) + String.valueOf(j), String.valueOf(i)); + } + input.put("MsgNum", String.valueOf(i)); + JSONEvent e = new JSONEvent(); + e.setHeaders(input); + e.setBody(String.valueOf(rand.nextGaussian()).getBytes("UTF-8")); + events.add(e); + } + Gson gson = new Gson(); + String json = gson.toJson(events, listType); + HttpsURLConnection httpsURLConnection = null; + Transaction transaction = null; + try { + TrustManager[] trustAllCerts = { + new X509TrustManager() { + @Override + public void checkClientTrusted(java.security.cert.X509Certificate[] x509Certificates, String s) + throws CertificateException { + // noop + } + + @Override + public void checkServerTrusted(java.security.cert.X509Certificate[] x509Certificates, String s) + throws CertificateException { + // noop + } + + public java.security.cert.X509Certificate[] getAcceptedIssuers() { + return null; + } + } + }; + + SSLContext sc = null; + javax.net.ssl.SSLSocketFactory factory = null; + if (System.getProperty("java.vendor").contains("IBM")) { + sc = SSLContext.getInstance("SSL_TLS"); + } else { + sc = SSLContext.getInstance("SSL"); + } + + HostnameVerifier hv = new HostnameVerifier() { + public boolean verify(String arg0, SSLSession arg1) { + return true; + } + }; + sc.init(null, trustAllCerts, new SecureRandom()); + + if (protocol != null) { + factory = new DisabledProtocolsSocketFactory(sc.getSocketFactory(), protocol); + } else { + factory = sc.getSocketFactory(); + } + HttpsURLConnection.setDefaultSSLSocketFactory(factory); + HttpsURLConnection.setDefaultHostnameVerifier(NoopHostnameVerifier.INSTANCE); + URL sslUrl = new URL("https://0.0.0.0:" + port); + httpsURLConnection = (HttpsURLConnection) sslUrl.openConnection(); + httpsURLConnection.setDoInput(true); + httpsURLConnection.setDoOutput(true); + httpsURLConnection.setRequestMethod("POST"); + httpsURLConnection.getOutputStream().write(json.getBytes()); + + int statusCode = httpsURLConnection.getResponseCode(); + Assert.assertEquals(200, statusCode); + + transaction = channel.getTransaction(); + transaction.begin(); + for (int i = 0; i < 10; i++) { + Event e = channel.take(); + Assert.assertNotNull(e); + Assert.assertEquals(String.valueOf(i), e.getHeaders().get("MsgNum")); + } + + } finally { + if (transaction != null) { + transaction.commit(); + transaction.close(); + } + httpsURLConnection.disconnect(); + } + } + + @Test + public void testHttpsSourceNonHttpsClient() throws Exception { + Type listType = new TypeToken<List<JSONEvent>>() {}.getType(); + List<JSONEvent> events = new ArrayList<JSONEvent>(); + Random rand = new Random(); + for (int i = 0; i < 10; i++) { + Map<String, String> input = Maps.newHashMap(); + for (int j = 0; j < 10; j++) { + input.put(String.valueOf(i) + String.valueOf(j), String.valueOf(i)); + } + input.put("MsgNum", String.valueOf(i)); + JSONEvent e = new JSONEvent(); + e.setHeaders(input); + e.setBody(String.valueOf(rand.nextGaussian()).getBytes("UTF-8")); + events.add(e); + } + Gson gson = new Gson(); + String json = gson.toJson(events, listType); + HttpURLConnection httpURLConnection = null; + try { + URL url = new URL("http://0.0.0.0:" + httpsPort); + httpURLConnection = (HttpURLConnection) url.openConnection(); + httpURLConnection.setDoInput(true); + httpURLConnection.setDoOutput(true); + httpURLConnection.setRequestMethod("POST"); + httpURLConnection.getOutputStream().write(json.getBytes()); + httpURLConnection.getResponseCode(); + + Assert.fail("HTTP Client cannot connect to HTTPS source"); + } catch (Exception exception) { + Assert.assertTrue("Exception expected", true); + } finally { + httpURLConnection.disconnect(); + } + } + + private void takeWithEncoding(String encoding, int n, List<JSONEvent> events) throws Exception { + Transaction tx = httpChannel.getTransaction(); + tx.begin(); + Event e = null; + int i = 0; + while (true) { + e = httpChannel.take(); + if (e == null) { + break; + } + Event current = events.get(i++); + Assert.assertEquals(new String(current.getBody(), encoding), new String(e.getBody(), encoding)); + Assert.assertEquals(current.getHeaders(), e.getHeaders()); + } + Assert.assertEquals(n, events.size()); + tx.commit(); + tx.close(); + } + + private void testBatchWithVariousEncoding(String encoding) throws Exception { + testBatchWithVariousEncoding(encoding, 50); + } + + private void testBatchWithVariousEncoding(String encoding, int n) throws Exception { + List<JSONEvent> events = putWithEncoding(encoding, n).events; + takeWithEncoding(encoding, n, events); + } + + private class ResultWrapper { + public final HttpResponse response; + public final List<JSONEvent> events; + + public ResultWrapper(HttpResponse resp, List<JSONEvent> events) { + this.response = resp; + this.events = events; + } + } + + private class DisabledProtocolsSocketFactory extends javax.net.ssl.SSLSocketFactory { + + private final javax.net.ssl.SSLSocketFactory socketFactory; + private final String[] protocols; + + DisabledProtocolsSocketFactory(javax.net.ssl.SSLSocketFactory factory, String protocol) { + this.socketFactory = factory; + protocols = new String[1]; + protocols[0] = protocol; + } + + @Override + public String[] getDefaultCipherSuites() { + return socketFactory.getDefaultCipherSuites(); + } + + @Override + public String[] getSupportedCipherSuites() { + return socketFactory.getSupportedCipherSuites(); + } + + @Override + public Socket createSocket(Socket socket, String s, int i, boolean b) throws IOException { + SSLSocket sc = (SSLSocket) socketFactory.createSocket(socket, s, i, b); + sc.setEnabledProtocols(protocols); + return sc; + } + + @Override + public Socket createSocket(String s, int i) throws IOException, UnknownHostException { + SSLSocket sc = (SSLSocket) socketFactory.createSocket(s, i); + sc.setEnabledProtocols(protocols); + return sc; + } + + @Override + public Socket createSocket(String s, int i, InetAddress inetAddress, int i2) + throws IOException, UnknownHostException { + SSLSocket sc = (SSLSocket) socketFactory.createSocket(s, i, inetAddress, i2); + sc.setEnabledProtocols(protocols); + return sc; + } + + @Override + public Socket createSocket(InetAddress inetAddress, int i) throws IOException { + SSLSocket sc = (SSLSocket) socketFactory.createSocket(inetAddress, i); + sc.setEnabledProtocols(protocols); + return sc; + } + + @Override + public Socket createSocket(InetAddress inetAddress, int i, InetAddress inetAddress2, int i2) + throws IOException { + SSLSocket sc = (SSLSocket) socketFactory.createSocket( + inetAddress, i, + inetAddress2, i2); + sc.setEnabledProtocols(protocols); + return sc; + } + } +}
diff --git a/src/test/java/org/apache/flume/http/source/TestJSONHandler.java b/src/test/java/org/apache/flume/http/source/TestJSONHandler.java new file mode 100644 index 0000000..ae97e36 --- /dev/null +++ b/src/test/java/org/apache/flume/http/source/TestJSONHandler.java
@@ -0,0 +1,238 @@ +/* + * 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.flume.http.source; + +import java.lang.reflect.Type; +import java.nio.charset.UnsupportedCharsetException; +import java.util.List; +import java.util.Map; +import java.util.Random; + +import org.apache.flume.Event; +import org.apache.flume.event.JSONEvent; +import org.junit.Before; +import org.junit.Test; + +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; + +import jakarta.servlet.http.HttpServletRequest; +import junit.framework.Assert; + +/** + * + */ +public class TestJSONHandler { + + HTTPSourceHandler handler; + + @Before + public void setUp() { + handler = new JSONHandler(); + } + + @Test + public void testMultipleEvents() throws Exception { + String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + e = deserialized.get(1); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testMultipleEventsUTF16() throws Exception { + String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "UTF-16"); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-16")); + e = deserialized.get(1); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-16")); + } + + @Test + public void testMultipleEventsUTF32() throws Exception { + String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "UTF-32"); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-32")); + e = deserialized.get(1); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-32")); + } + + @Test + public void testMultipleEventsUTF8() throws Exception { + String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "UTF-8"); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + e = deserialized.get(1); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testEscapedJSON() throws Exception { + // JSON allows escaping double quotes to add it in the data. + String json = "[{\"headers\":{\"a\": \"b\"}}," + "{\"headers\":{\"e\": \"f\"},\"body\": \"rand\\\"om_body2\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertTrue(e.getBody().length == 0); + e = deserialized.get(1); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("rand\"om_body2", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testNoBody() throws Exception { + String json = "[{\"headers\" : {\"a\": \"b\"}}," + "{\"headers\" : {\"e\": \"f\"},\"body\": \"random_body2\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertTrue(e.getBody().length == 0); + e = deserialized.get(1); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testSingleHTMLEvent() throws Exception { + String json = "[{\"headers\": {\"a\": \"b\"}," + "\"body\": \"<html><body>test</body></html>\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("<html><body>test</body></html>", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testSingleEvent() throws Exception { + String json = "[{\"headers\" : {\"a\": \"b\"},\"body\": \"random_body\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + } + + @Test(expected = HTTPBadRequestException.class) + public void testBadEvent() throws Exception { + String json = "{[\"a\": \"b\"],\"body\": \"random_body\"}"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + handler.getEvents(req); + Assert.fail(); + } + + @Test(expected = UnsupportedCharsetException.class) + public void testError() throws Exception { + String json = "[{\"headers\" : {\"a\": \"b\"},\"body\": \"random_body\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "ISO-8859-1"); + handler.getEvents(req); + Assert.fail(); + } + + @Test + public void testSingleEventInArray() throws Exception { + String json = "[{\"headers\": {\"a\": \"b\"},\"body\": \"random_body\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testMultipleLargeEvents() throws Exception { + String json = "[{\"headers\" : {\"a\": \"b\", \"a2\": \"b2\"," + + "\"a3\": \"b3\",\"a4\": \"b4\"},\"body\": \"random_body\"}," + + "{\"headers\" :{\"e\": \"f\",\"e2\": \"f2\"," + + "\"e3\": \"f3\",\"e4\": \"f4\",\"e5\": \"f5\"}," + + "\"body\": \"random_body2\"}," + + "{\"headers\" :{\"q1\": \"b\",\"q2\": \"b2\",\"q3\": \"b3\",\"q4\": \"b4\"}," + + "\"body\": \"random_bodyq\"}]"; + HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); + List<Event> deserialized = handler.getEvents(req); + Event e = deserialized.get(0); + Assert.assertNotNull(e); + Assert.assertEquals("b", e.getHeaders().get("a")); + Assert.assertEquals("b2", e.getHeaders().get("a2")); + Assert.assertEquals("b3", e.getHeaders().get("a3")); + Assert.assertEquals("b4", e.getHeaders().get("a4")); + Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); + e = deserialized.get(1); + Assert.assertNotNull(e); + Assert.assertEquals("f", e.getHeaders().get("e")); + Assert.assertEquals("f2", e.getHeaders().get("e2")); + Assert.assertEquals("f3", e.getHeaders().get("e3")); + Assert.assertEquals("f4", e.getHeaders().get("e4")); + Assert.assertEquals("f5", e.getHeaders().get("e5")); + Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-8")); + e = deserialized.get(2); + Assert.assertNotNull(e); + Assert.assertEquals("b", e.getHeaders().get("q1")); + Assert.assertEquals("b2", e.getHeaders().get("q2")); + Assert.assertEquals("b3", e.getHeaders().get("q3")); + Assert.assertEquals("b4", e.getHeaders().get("q4")); + Assert.assertEquals("random_bodyq", new String(e.getBody(), "UTF-8")); + } + + @Test + public void testDeserializarion() throws Exception { + Type listType = new TypeToken<List<JSONEvent>>() {}.getType(); + List<JSONEvent> events = Lists.newArrayList(); + Random rand = new Random(); + for (int i = 1; i < 10; i++) { + Map<String, String> input = Maps.newHashMap(); + for (int j = 1; j < 10; j++) { + input.put(String.valueOf(i) + String.valueOf(j), String.valueOf(i)); + } + JSONEvent e = new JSONEvent(); + e.setBody(String.valueOf(rand.nextGaussian()).getBytes("UTF-8")); + e.setHeaders(input); + events.add(e); + } + Gson gson = new Gson(); + List<Event> deserialized = handler.getEvents(new FlumeHttpServletRequestWrapper(gson.toJson(events, listType))); + int i = 0; + for (Event e : deserialized) { + Event current = events.get(i++); + Assert.assertEquals(new String(current.getBody(), "UTF-8"), new String(e.getBody(), "UTF-8")); + Assert.assertEquals(current.getHeaders(), e.getHeaders()); + } + } +}