You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2015/02/17 20:16:50 UTC
[1/5] incubator-streams git commit: create BroadcastMessagePersister
when BroadcastURI is provided.
Repository: incubator-streams
Updated Branches:
refs/heads/master 18d14f010 -> f471d784e
create BroadcastMessagePersister when BroadcastURI is provided.
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/b65e198d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/b65e198d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/b65e198d
Branch: refs/heads/master
Commit: b65e198dba3a97535b91afe1c9898af23426911b
Parents: c775391
Author: sblackmon <sb...@w2odigital.com>
Authored: Wed Feb 11 18:07:03 2015 -0600
Committer: sblackmon <sb...@w2odigital.com>
Committed: Wed Feb 11 18:07:03 2015 -0600
----------------------------------------------------------------------
.../streams/monitoring/tasks/BroadcastMonitorThread.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/b65e198d/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
----------------------------------------------------------------------
diff --git a/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java b/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
index ea7edd1..6867bd3 100644
--- a/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
+++ b/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
@@ -20,6 +20,7 @@ package org.apache.streams.monitoring.tasks;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.streams.jackson.*;
import org.apache.streams.monitoring.persist.MessagePersister;
@@ -58,11 +59,13 @@ public class BroadcastMonitorThread extends NotificationBroadcasterSupport imple
server = ManagementFactory.getPlatformMBeanServer();
-
setBroadcastURI();
setWaitTime();
- messagePersister = new SLF4JMessagePersister();
+ if( !Strings.isNullOrEmpty(broadcastURI))
+ messagePersister = new BroadcastMessagePersister(broadcastURI);
+ else
+ messagePersister = new SLF4JMessagePersister();
initializeObjectMapper();
[2/5] incubator-streams git commit: added working UDP log stash
persister
Posted by sb...@apache.org.
added working UDP log stash persister
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/5667a7e6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/5667a7e6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/5667a7e6
Branch: refs/heads/master
Commit: 5667a7e6a59038060d38c37a15d4c8a30eb11fab
Parents: b65e198
Author: sblackmon <sb...@apache.org>
Authored: Thu Feb 12 14:47:13 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Thu Feb 12 18:17:19 2015 -0600
----------------------------------------------------------------------
.../impl/LogstashUdpMessagePersister.java | 104 +++++++++++++++++++
.../tasks/BroadcastMonitorThread.java | 21 ++--
2 files changed, 119 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/5667a7e6/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
----------------------------------------------------------------------
diff --git a/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java b/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
new file mode 100644
index 0000000..6e145b9
--- /dev/null
+++ b/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
@@ -0,0 +1,104 @@
+/*
+ * 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
+ *
+ * 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.streams.monitoring.persist.impl;
+
+import org.apache.streams.monitoring.persist.MessagePersister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.Executors;
+
+public class LogstashUdpMessagePersister implements MessagePersister {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(LogstashUdpMessagePersister.class);
+ private String broadcastURI;
+ URI uri;
+
+ public LogstashUdpMessagePersister(String broadcastURI) {
+ this.broadcastURI = broadcastURI;
+ setup();
+ }
+
+ public void setup() {
+
+ try {
+ uri = new URI(broadcastURI);
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ }
+
+ }
+ @Override
+ /**
+ * Given a list of messages as Strings, broadcast them to the broadcastURI
+ * (if one is defined)
+ * @param messages
+ * @return int status code from POST response
+ */
+ public int persistMessages(List<String> messages) {
+ int responseCode = -1;
+
+ if(broadcastURI != null) {
+ DatagramSocket socket = null;
+ try {
+ socket = new DatagramSocket();
+ } catch (SocketException e) {
+ LOGGER.error("Metrics Broadcast Setup Failed: " + e.getMessage());
+ }
+ try {
+ ByteBuffer toWrite = ByteBuffer.wrap(serializeMessages(messages).getBytes());
+ byte[] byteArray = toWrite.array();
+ DatagramPacket packet = new DatagramPacket(byteArray, byteArray.length);
+ socket.connect(new InetSocketAddress(uri.getHost(), uri.getPort()));
+ socket.send(packet);
+ } catch( Exception e ) {
+ LOGGER.error("Metrics Broadcast Failed: " + e.getMessage());
+ } finally {
+ socket.close();
+ }
+ }
+
+ return responseCode;
+ }
+
+ /**
+ * Given a List of String messages, convert them to a JSON array
+ * @param messages
+ * @return Serialized version of this JSON array
+ */
+ private String serializeMessages(List<String> messages) {
+
+ StringBuilder json_lines = new StringBuilder();
+ for(String message : messages) {
+ json_lines.append(message).append('\n');
+ }
+
+ return json_lines.toString();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/5667a7e6/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
----------------------------------------------------------------------
diff --git a/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java b/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
index 6867bd3..948863e 100644
--- a/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
+++ b/streams-monitoring/src/main/java/org/apache/streams/monitoring/tasks/BroadcastMonitorThread.java
@@ -20,17 +20,19 @@ package org.apache.streams.monitoring.tasks;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
-import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.apache.streams.jackson.*;
import org.apache.streams.monitoring.persist.MessagePersister;
import org.apache.streams.monitoring.persist.impl.BroadcastMessagePersister;
+import org.apache.streams.monitoring.persist.impl.LogstashUdpMessagePersister;
import org.apache.streams.monitoring.persist.impl.SLF4JMessagePersister;
import org.apache.streams.pojo.json.*;
import org.slf4j.Logger;
import javax.management.*;
import java.lang.management.ManagementFactory;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -47,7 +49,7 @@ public class BroadcastMonitorThread extends NotificationBroadcasterSupport imple
private long waitTime;
private ObjectMapper objectMapper;
private Map<String, Object> streamConfig;
- private String broadcastURI = null;
+ private URI broadcastURI = null;
private MessagePersister messagePersister;
private volatile boolean keepRunning;
@@ -62,8 +64,11 @@ public class BroadcastMonitorThread extends NotificationBroadcasterSupport imple
setBroadcastURI();
setWaitTime();
- if( !Strings.isNullOrEmpty(broadcastURI))
- messagePersister = new BroadcastMessagePersister(broadcastURI);
+ if( broadcastURI != null )
+ if( broadcastURI.getScheme().equals("http"))
+ messagePersister = new BroadcastMessagePersister(broadcastURI.toString());
+ else if( broadcastURI.getScheme().equals("udp"))
+ messagePersister = new LogstashUdpMessagePersister(broadcastURI.toString());
else
messagePersister = new SLF4JMessagePersister();
@@ -143,7 +148,11 @@ public class BroadcastMonitorThread extends NotificationBroadcasterSupport imple
streamConfig.containsKey("broadcastURI") &&
streamConfig.get("broadcastURI") != null &&
streamConfig.get("broadcastURI") instanceof String) {
- broadcastURI = streamConfig.get("broadcastURI").toString();
+ try {
+ broadcastURI = new URI(streamConfig.get("broadcastURI").toString());
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ }
}
}
@@ -177,7 +186,7 @@ public class BroadcastMonitorThread extends NotificationBroadcasterSupport imple
}
public String getBroadcastURI() {
- return broadcastURI;
+ return broadcastURI.toString();
}
public long getWaitTime() {
[5/5] incubator-streams git commit: Merge branch 'STREAMS-283' into
asf-master
Posted by sb...@apache.org.
Merge branch 'STREAMS-283' into asf-master
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/f471d784
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/f471d784
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/f471d784
Branch: refs/heads/master
Commit: f471d784e27a4703b785bbc87518b4ace30a2759
Parents: 18d14f0 7294fa6
Author: sblackmon <sb...@apache.org>
Authored: Tue Feb 17 12:51:46 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 17 12:51:46 2015 -0600
----------------------------------------------------------------------
.../impl/LogstashUdpMessagePersister.java | 104 +++++++++++++++++++
.../tasks/BroadcastMonitorThread.java | 22 +++-
.../impl/LogstashUdpMessagePersisterTest.java | 75 +++++++++++++
3 files changed, 196 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
[4/5] incubator-streams git commit: addressed feedback from
https://github.com/apache/incubator-streams/pull/183
Posted by sb...@apache.org.
addressed feedback from https://github.com/apache/incubator-streams/pull/183
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/7294fa64
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/7294fa64
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/7294fa64
Branch: refs/heads/master
Commit: 7294fa6429823c1fd3bfff460acea32a8a1e9ab9
Parents: db6400b
Author: sblackmon <sb...@apache.org>
Authored: Tue Feb 17 12:28:21 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 17 12:28:21 2015 -0600
----------------------------------------------------------------------
.../monitoring/persist/impl/LogstashUdpMessagePersister.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/7294fa64/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
----------------------------------------------------------------------
diff --git a/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java b/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
index 6e145b9..312502c 100644
--- a/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
+++ b/streams-monitoring/src/main/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersister.java
@@ -49,7 +49,7 @@ public class LogstashUdpMessagePersister implements MessagePersister {
try {
uri = new URI(broadcastURI);
} catch (URISyntaxException e) {
- e.printStackTrace();
+ LOGGER.error(e.getMessage());
}
}
[3/5] incubator-streams git commit: added test for UdpMessagePersister
Posted by sb...@apache.org.
added test for UdpMessagePersister
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/db6400b5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/db6400b5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/db6400b5
Branch: refs/heads/master
Commit: db6400b517f80acf4b904d824b619e05f0d5fae3
Parents: 5667a7e
Author: sblackmon <sb...@apache.org>
Authored: Fri Feb 13 09:37:41 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Fri Feb 13 09:38:19 2015 -0600
----------------------------------------------------------------------
.../impl/LogstashUdpMessagePersisterTest.java | 75 ++++++++++++++++++++
1 file changed, 75 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/db6400b5/streams-monitoring/src/test/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersisterTest.java
----------------------------------------------------------------------
diff --git a/streams-monitoring/src/test/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersisterTest.java b/streams-monitoring/src/test/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersisterTest.java
new file mode 100644
index 0000000..faa99a2
--- /dev/null
+++ b/streams-monitoring/src/test/java/org/apache/streams/monitoring/persist/impl/LogstashUdpMessagePersisterTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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
+ *
+ * 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.streams.monitoring.persist.impl;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketException;
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+public class LogstashUdpMessagePersisterTest {
+
+ private static final Logger LOGGER = org.slf4j.LoggerFactory.getLogger(LogstashUdpMessagePersisterTest.class);
+
+ DatagramSocket socket = null;
+
+ @Before
+ public void setup() {
+ try {
+ socket = new DatagramSocket(56789);
+ } catch (SocketException e) {
+ LOGGER.error("Metrics Broadcast Test Setup Failed: " + e.getMessage());
+ }
+ }
+
+
+ @Test
+ public void testFailedPersist() {
+ LogstashUdpMessagePersister persister = new LogstashUdpMessagePersister("udp://127.0.0.1:56789");
+
+ List<String> messageArray = Lists.newArrayList();
+ for(int x = 0; x < 10; x ++) {
+ messageArray.add("Fake_message #" + x);
+ }
+
+ persister.persistMessages(messageArray);
+ byte[] receiveData = new byte[1024];
+
+ DatagramPacket messageDatagram = new DatagramPacket(receiveData, receiveData.length);
+
+ try {
+ socket.receive(messageDatagram);
+ assertNotNull(messageDatagram);
+ List<String> messages = Lists.newArrayList(Splitter.on('\n').split(new String(messageDatagram.getData())));
+ assertEquals(messageArray, messages.subList(0,10));
+ } catch (IOException e) {
+ LOGGER.error("Metrics Broadcast Test Failed: " + e.getMessage());
+ }
+
+ }
+
+}