Adding type of stream processing disblaed notification

This notification is shown, when a stream has been disabled due to
repeated processing timeouts during stream rule matching. This allows
the user to check the configured stream rules, correct them and re-
enable the stream.

Author: dennisoelkers

app/controllers/api/SystemApiController.java

@@ -21,12 +21,14 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
 import com.google.inject.Inject;
 import controllers.AuthenticatedController;
 import lib.APIException;
 import lib.metrics.Meter;
 import models.*;
 import play.libs.F;
+import play.libs.Json;
 import play.mvc.Http;
 import play.mvc.Result;
 
@@ -84,6 +86,7 @@ public Result notifications() {
             result.put("count", clusterService.allNotifications().size());*/
             List<Notification> notifications = clusterService.allNotifications();
 
+
             return ok(new Gson().toJson(notifications)).as("application/json");
         } catch (IOException e) {
             return internalServerError("io exception");

app/lib/notifications/InputFailedToStartNotification.java

@@ -16,9 +16,9 @@ public class InputFailedToStartNotification implements NotificationType {
     private final String DESCRIPTION;
 
     public InputFailedToStartNotification(Notification notification) {
-        DESCRIPTION = "Input " + (String)notification.getDetail("input_id") + " has failed to start on node " +
+        DESCRIPTION = "Input " + notification.getDetail("input_id") + " has failed to start on node " +
                 notification.getNodeId() + " for this reason: \"" +
-                (String)notification.getDetail("reason") + "\". " +
+                notification.getDetail("reason") + "\". " +
                 "This means that you are unable to receive any messages from this input. This is mostly " +
                 "an indication for a misconfiguration or an error. " + "" +
                 "You can click <a href='" + routes.InputsController.index() + "'>here</a> to solve this";

app/lib/notifications/StreamProcessingDisabled.java

@@ -0,0 +1,44 @@
+package lib.notifications;
+
+import com.google.common.collect.Maps;
+import models.Stream;
+import models.SystemJob;
+
+import java.util.Map;
+
+/**
+ * @author Dennis Oelkers <[email protected]>
+ */
+public class StreamProcessingDisabled implements NotificationType {
+    private final String streamTitle;
+    private final long faultCount;
+
+    public StreamProcessingDisabled(String streamTitle, long faultCount) {
+        this.streamTitle = streamTitle;
+        this.faultCount = faultCount;
+    }
+
+    @Override
+    public Map<SystemJob.Type, String> options() {
+        return Maps.newHashMap();
+    }
+
+    @Override
+    public String getTitle() {
+        return "Processing of a stream has been disabled due to excessive processing time.";
+    }
+
+    @Override
+    public String getDescription() {
+        return "The processing of stream <em>" + streamTitle
+                + "</em> has taken too long for " + faultCount + " times. "
+                + "To protect the stability of message processing, this stream has been disabled. "
+                + "Please correct the stream rules and reenable the stream.";
+
+    }
+
+    @Override
+    public boolean isCloseable() {
+        return true;
+    }
+}

app/models/ClusterService.java

@@ -50,16 +50,19 @@ public class ClusterService {
     private final SystemJob.Factory systemJobFactory;
     private final ServerNodes serverNodes;
     private final NodeService nodeService;
+    private final StreamService streamService;
 
     @Inject
     private ClusterService(ApiClient api,
                            SystemJob.Factory systemJobFactory,
                            ServerNodes serverNodes,
-                           NodeService nodeService) {
+                           NodeService nodeService,
+                           StreamService streamService) {
         this.api = api;
         this.systemJobFactory = systemJobFactory;
         this.serverNodes = serverNodes;
         this.nodeService = nodeService;
+        this.streamService = streamService;
     }
 
     public void triggerSystemJob(SystemJob.Type type, User user) throws IOException, APIException {
@@ -76,7 +79,7 @@ public List<Notification> allNotifications() throws IOException, APIException {
         List<Notification> notifications = Lists.newArrayList();
         for (NotificationSummaryResponse notification : r.notifications) {
             try {
-                notifications.add(new Notification(notification));
+                notifications.add(new Notification(notification, streamService, nodeService));
             } catch(IllegalArgumentException e) {
                 play.Logger.warn("There is a notification type we can't handle: [" + notification.type + "]");
                 continue;

app/models/Notification.java

@@ -18,20 +18,21 @@
  */
 package models;
 
-import com.google.inject.Inject;
+import lib.APIException;
 import lib.notifications.*;
 import models.api.responses.system.NotificationSummaryResponse;
 import org.joda.time.DateTime;
 
+import java.io.IOException;
 import java.util.Map;
 
 /**
  * @author Lennart Koopmann <[email protected]>
  */
 public class Notification {
 
-    @Inject
-    private NodeService nodeService;
+    private transient final NodeService nodeService;
+    private transient final StreamService streamService;
 
     public enum Type {
         DEFLECTOR_EXISTS_AS_INDEX,
@@ -41,7 +42,8 @@ public enum Type {
         NO_INPUT_RUNNING,
         INPUT_FAILED_TO_START,
         CHECK_SERVER_CLOCKS,
-        OUTDATED_VERSION;
+        OUTDATED_VERSION,
+        STREAM_PROCESSING_DISABLED;
 
         public static Type fromString(String name) {
             return valueOf(name.toUpperCase());
@@ -58,15 +60,17 @@ public enum Severity {
     private final String node_id;
     private final Map<String, Object> details;
 
-    public Notification(NotificationSummaryResponse x) {
+    public Notification(NotificationSummaryResponse x, StreamService streamService, NodeService nodeService) {
+        this.streamService = streamService;
+        this.nodeService = nodeService;
         this.type = Type.valueOf(x.type.toUpperCase());
         this.timestamp = DateTime.parse(x.timestamp);
         this.severity = Severity.valueOf(x.severity.toUpperCase());
         this.node_id = x.node_id;
         this.details = x.details;
     }
 
-    public NotificationType get() {
+    public NotificationType get() throws IOException, APIException {
         switch (type) {
             case DEFLECTOR_EXISTS_AS_INDEX:
                 return new DeflectorExistsAsIndexNotification();
@@ -84,6 +88,10 @@ public NotificationType get() {
                 return new CheckServerClocksNotification();
             case OUTDATED_VERSION:
                 return new OutdatedVersionNotification(this);
+            case STREAM_PROCESSING_DISABLED:
+                Stream stream = streamService.get(details.get("stream_id").toString());
+                long faultCount = Math.round((double)details.get("fault_count"));
+                return new StreamProcessingDisabled(stream.getTitle(), faultCount);
         }
 
         throw new RuntimeException("No notification registered for " + type);