You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@karaf.apache.org by cs...@apache.org on 2016/06/23 16:04:39 UTC
karaf-decanter git commit: [KARAF-4594] Log4JSocketCollector multiple
clients
Repository: karaf-decanter
Updated Branches:
refs/heads/master 40259179d -> c73ad8a2e
[KARAF-4594] Log4JSocketCollector multiple clients
This commit fixes current SocketCollector limitation : it handled only 1 client
simultaneously.
With the following implementation :
* you configure number of workers
* serverSocket is handled by a master thread
* client sockets are handled by the workers
I've also added some unit tests, but since I'm testing a multi-threaded module results are unpredictable (I've set unit test timeout to 10s so it should be ok to process 1 ou 2 events in most cases).
Project: http://git-wip-us.apache.org/repos/asf/karaf-decanter/repo
Commit: http://git-wip-us.apache.org/repos/asf/karaf-decanter/commit/c73ad8a2
Tree: http://git-wip-us.apache.org/repos/asf/karaf-decanter/tree/c73ad8a2
Diff: http://git-wip-us.apache.org/repos/asf/karaf-decanter/diff/c73ad8a2
Branch: refs/heads/master
Commit: c73ad8a2e1d529c76e7f86ef962b4a193a2bb239
Parents: 4025917
Author: Adrian Gonzalez <ag...@talend.com>
Authored: Thu Jun 23 17:10:57 2016 +0200
Committer: Christian Schneider <ch...@die-schneider.net>
Committed: Thu Jun 23 18:02:57 2016 +0200
----------------------------------------------------------------------
...ache.karaf.decanter.collector.log.socket.cfg | 1 +
.../collector/log/socket/SocketCollector.java | 73 +++++--
.../log/socket/SocketCollectorTest.java | 215 ++++++++++++++++++-
.../collector/log/socket/SocketUtils.java | 51 +++++
4 files changed, 320 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/karaf-decanter/blob/c73ad8a2/collector/log4j-socket/src/main/cfg/org.apache.karaf.decanter.collector.log.socket.cfg
----------------------------------------------------------------------
diff --git a/collector/log4j-socket/src/main/cfg/org.apache.karaf.decanter.collector.log.socket.cfg b/collector/log4j-socket/src/main/cfg/org.apache.karaf.decanter.collector.log.socket.cfg
index 29e4dcb..b93137d 100644
--- a/collector/log4j-socket/src/main/cfg/org.apache.karaf.decanter.collector.log.socket.cfg
+++ b/collector/log4j-socket/src/main/cfg/org.apache.karaf.decanter.collector.log.socket.cfg
@@ -3,3 +3,4 @@
#
#port=4560
+#workers=10
http://git-wip-us.apache.org/repos/asf/karaf-decanter/blob/c73ad8a2/collector/log4j-socket/src/main/java/org/apache/karaf/decanter/collector/log/socket/SocketCollector.java
----------------------------------------------------------------------
diff --git a/collector/log4j-socket/src/main/java/org/apache/karaf/decanter/collector/log/socket/SocketCollector.java b/collector/log4j-socket/src/main/java/org/apache/karaf/decanter/collector/log/socket/SocketCollector.java
index 145c5bb..c080d57 100644
--- a/collector/log4j-socket/src/main/java/org/apache/karaf/decanter/collector/log/socket/SocketCollector.java
+++ b/collector/log4j-socket/src/main/java/org/apache/karaf/decanter/collector/log/socket/SocketCollector.java
@@ -48,6 +48,8 @@ import org.slf4j.LoggerFactory;
@Component(name = "org.apache.karaf.decanter.collector.log.socket", immediate = true)
public class SocketCollector implements Closeable, Runnable {
+ public static final String PORT_NAME = "port";
+ public static final String WORKERS_NAME = "workers";
private static final Logger LOGGER = LoggerFactory.getLogger(SocketCollector.class);
private ServerSocket serverSocket;
private EventAdmin eventAdmin;
@@ -59,10 +61,12 @@ public class SocketCollector implements Closeable, Runnable {
@Activate
public void activate(ComponentContext context) throws IOException {
this.properties = context.getProperties();
- int port = Integer.parseInt(getProperty(this.properties, "port", "4560"));
+ int port = Integer.parseInt(getProperty(this.properties, PORT_NAME, "4560"));
+ int workers = Integer.parseInt(getProperty(this.properties, WORKERS_NAME, "10"));
LOGGER.info("Starting Log4j Socket collector on port {}", port);
this.serverSocket = new ServerSocket(port);
- this.executor = Executors.newFixedThreadPool(1);
+ // adding 1 for serverSocket handling
+ this.executor = Executors.newFixedThreadPool(workers + 1);
this.executor.execute(this);
this.open = true;
}
@@ -74,29 +78,30 @@ public class SocketCollector implements Closeable, Runnable {
@Override
public void run() {
while (open) {
- try ( //
+ try {
Socket socket = serverSocket.accept();
- ObjectInputStream ois = new ObjectInputStream(new BufferedInputStream(socket
- .getInputStream())); //
- ) {
- while (open) {
- try {
- Object event = ois.readObject();
- if (event instanceof LoggingEvent) {
- handleLog4j((LoggingEvent)event);
- }
- } catch (ClassNotFoundException e) {
- LOGGER.warn("Unable to deserialize event from " + socket.getInetAddress(), e);
- }
- }
- } catch (EOFException e) {
- LOGGER.debug("Log client closed the connection.", e);
+ LOGGER.debug("Connected to client at {}", socket.getInetAddress());
+ this.executor.execute(new SocketRunnable(socket));
} catch (IOException e) {
LOGGER.warn("Exception receiving log.", e);
}
}
}
+ /**
+ * Returns true if this class is active (waiting for client logging events)
+ */
+ boolean isOpen() {
+ return open;
+ }
+
+ /**
+ * returns executor service used for serverSocket and client socket handling
+ */
+ ExecutorService getExecutorService() {
+ return executor;
+ }
+
private void handleLog4j(LoggingEvent event) throws UnknownHostException {
LOGGER.debug("Received log event {}", event.getLoggerName());
Map<String, Object> data = new HashMap<>();
@@ -192,4 +197,36 @@ public class SocketCollector implements Closeable, Runnable {
this.eventAdmin = eventAdmin;
}
+ private class SocketRunnable implements Runnable {
+ private Socket clientSocket;
+
+ public SocketRunnable(Socket clientSocket) {
+ this.clientSocket = clientSocket;
+ }
+
+ public void run() {
+ try (ObjectInputStream ois = new ObjectInputStream(new BufferedInputStream(clientSocket
+ .getInputStream()))) {
+ while (open) {
+ try {
+ Object event = ois.readObject();
+ if (event instanceof LoggingEvent) {
+ handleLog4j((LoggingEvent)event);
+ }
+ } catch (ClassNotFoundException e) {
+ LOGGER.warn("Unable to deserialize event from " + clientSocket.getInetAddress(), e);
+ }
+ }
+ } catch (EOFException e) {
+ LOGGER.debug("Log client closed the connection.", e);
+ } catch (IOException e) {
+ LOGGER.warn("Exception receiving log.", e);
+ }
+ try {
+ clientSocket.close();
+ } catch (IOException e) {
+ LOGGER.info("Error closing socket", e);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/karaf-decanter/blob/c73ad8a2/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketCollectorTest.java
----------------------------------------------------------------------
diff --git a/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketCollectorTest.java b/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketCollectorTest.java
index b20bd08..0365e2d 100644
--- a/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketCollectorTest.java
+++ b/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketCollectorTest.java
@@ -16,11 +16,58 @@
*/
package org.apache.karaf.decanter.collector.log.socket;
-import org.apache.karaf.decanter.collector.log.socket.SocketCollector;
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Dictionary;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.log4j.Level;
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.log4j.spi.NOPLogger;
+import org.apache.log4j.spi.NOPLoggerRepository;
+import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
+import org.osgi.framework.Bundle;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceReference;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.ComponentInstance;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+
+public class SocketCollectorTest {
+
+ private ComponentContext componentContext;
+ private SocketCollector collector;
+ private int port;
+ private EventAdminStub eventAdmin;
+
+ @Before
+ public void setUp() throws IOException {
+ port = SocketUtils.findAvailablePort();
+ eventAdmin = new EventAdminStub();
+ collector = new SocketCollector();
+ collector.setEventAdmin(eventAdmin);
+ componentContext = new ComponentContextStub();
+ componentContext.getProperties().put(SocketCollector.PORT_NAME, String.valueOf(port));
+ componentContext.getProperties().put(SocketCollector.WORKERS_NAME, "1");
+ }
-public class SocketCollectorTest {
+ @After
+ public void tearDown() throws IOException {
+ if (collector != null && collector.isOpen()) {
+ collector.close();
+ }
+ }
@Test
public void testLoggerName2Topic() {
@@ -28,4 +75,168 @@ public class SocketCollectorTest {
Assert.assertEquals("decanter/collect/log/test/Tomcat/localhost", topic);
}
+ /**
+ * Test event handling (1 event)
+ */
+ @Test
+ public void testSocket() throws Exception {
+ activate();
+ sendEventOnSocket(newLoggingEvent("Sample message"));
+ waitUntilEventCountHandled(1);
+ assertEquals("Event(s) should have been correctly handled", 1, eventAdmin.getPostEvents().size());
+ }
+
+ /**
+ * Test event handling with multiple clients
+ */
+ @Test
+ public void testSocketMultipleClients() throws Exception {
+
+ activate();
+
+ Runnable runnable = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ sendEventOnSocket(newLoggingEvent("Sample message"));
+ } catch (Throwable e) {
+ e.printStackTrace();
+ }
+ }
+ };
+
+ // starts 2 client threads
+ Thread thread;
+ thread = new Thread(runnable);
+ thread.start();
+ thread = new Thread(runnable);
+ thread.start();
+
+ waitUntilEventCountHandled(2);
+ assertEquals("Event(s) should have been correctly handled", 2, eventAdmin.getPostEvents().size());
+ }
+
+ private void waitUntilEventCountHandled(int eventCount) throws InterruptedException {
+ long timeout = 20000L;
+ long start = System.currentTimeMillis();
+ boolean hasTimeoutReached = false;
+ do {
+ hasTimeoutReached = ((System.currentTimeMillis() - start) > timeout);
+ Thread.sleep(10L);
+ } while (eventAdmin.getPostEvents().size() < eventCount && hasTimeoutReached == false);
+ }
+
+ private LoggingEvent newLoggingEvent(String message) {
+ return new LoggingEvent(this.getClass().getName(), new NOPLogger(new NOPLoggerRepository(), "NOP"),
+ System.currentTimeMillis(), Level.INFO, message,
+ Thread.currentThread().getName(), null, null, null, new Properties());
+ }
+
+ /**
+ * Launches serverSocket on available port
+ *
+ * @throws InterruptedException
+ */
+ private void activate() throws IOException, InterruptedException {
+ collector.activate(componentContext);
+ Thread.sleep(200L);
+ }
+
+ private void sendEventOnSocket(LoggingEvent event) throws IOException {
+ try (Socket socket = new Socket()) {
+ socket.connect(new InetSocketAddress("localhost", port), 5000);
+ try (ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream())) {
+ out.writeObject(event);
+ out.flush();
+ }
+ }
+ }
+
+ /**
+ * Stub used only for this unit test
+ */
+ private static class ComponentContextStub implements ComponentContext {
+
+ private Dictionary properties = new Hashtable<>();
+
+ @Override
+ public Dictionary getProperties() {
+ return properties;
+ }
+
+ @Override
+ public Object locateService(String name) {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public Object locateService(String name, ServiceReference reference) {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public Object[] locateServices(String name) {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public BundleContext getBundleContext() {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public Bundle getUsingBundle() {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public ComponentInstance getComponentInstance() {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public void enableComponent(String name) {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public void disableComponent(String name) {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ @Override
+ public ServiceReference getServiceReference() {
+ throw new NoSuchMethodError("Unimplemented method");
+ }
+
+ }
+
+ private static class EventAdminStub implements EventAdmin {
+ private List<Event> postEvents = new ArrayList<>();
+ private List<Event> sendEvents = new ArrayList<>();
+
+ @Override
+ public void postEvent(Event event) {
+ postEvents.add(event);
+ }
+
+ @Override
+ public void sendEvent(Event event) {
+ sendEvents.add(event);
+ }
+
+ public List<Event> getPostEvents() {
+ return postEvents;
+ }
+
+ public List<Event> getSendEvents() {
+ return sendEvents;
+ }
+
+ public void reset() {
+ postEvents.clear();
+ sendEvents.clear();
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/karaf-decanter/blob/c73ad8a2/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketUtils.java
----------------------------------------------------------------------
diff --git a/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketUtils.java b/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketUtils.java
new file mode 100644
index 0000000..b2950fc
--- /dev/null
+++ b/collector/log4j-socket/src/test/java/org/apache/karaf/decanter/collector/log/socket/SocketUtils.java
@@ -0,0 +1,51 @@
+package org.apache.karaf.decanter.collector.log.socket;
+
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.util.Random;
+
+import javax.net.ServerSocketFactory;
+
+/**
+ * Utility class to determine if TCP port is available on localhost.
+ *
+ * @author agonzalez
+ */
+public final class SocketUtils {
+
+ private static final Random random = new Random();
+ public static final int MIN_PORT = 40000;
+ public static final int MAX_PORT = 65535;
+
+ /**
+ * Returns the first available port between {@link #MIN_PORT} and
+ * {@link #MAX_PORT}
+ */
+ public static int findAvailablePort() {
+ int port;
+ do {
+ port = findRandomPort(MIN_PORT, MAX_PORT);
+
+ } while (!isPortAvailable(port));
+ return port;
+ }
+
+ private static boolean isPortAvailable(int port) {
+ try {
+ ServerSocket serverSocker = ServerSocketFactory.getDefault().createServerSocket(port, 1,
+ InetAddress.getByName("localhost"));
+ serverSocker.close();
+ return true;
+ } catch (Exception ex) {
+ return false;
+ }
+ }
+
+ private static int findRandomPort(int minPort, int maxPort) {
+ if (maxPort < minPort) {
+ throw new IllegalArgumentException("maxPort should be >= minPort");
+ }
+ int portRange = maxPort - minPort + 1;
+ return random.nextInt(portRange) + minPort;
+ }
+}