Added support for historic log search. Added docker and lostash confi…

…guraton to the distribution to begin tracking changes with releases.
lmco · Feb 23, 2015 · bd2381d · bd2381d
1 parent f188d10
commit bd2381d
Show file tree

Hide file tree

Showing 11 changed files with 133 additions and 32 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,5 +1,6 @@
 target/
 nb-configuration.xml
+nbactions.xml
 .DS_Store
 *.DS_Store
 *.iml

diff --git a/pom.xml b/pom.xml
@@ -48,7 +48,7 @@
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
         <!-- Dependency Versions -->
-        <storm.version>0.9.1-incubating</storm.version>
+        <storm.version>0.9.3</storm.version>
         <jetty.version>9.0.0.RC2</jetty.version>
         <jackson.version>2.1.5</jackson.version>
         <jersey.version>1.17.1</jersey.version>
@@ -409,16 +409,6 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
-        <!--
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>log4j</groupId>
-            <artifactId>log4j</artifactId>
-        </dependency>
-        -->
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

diff --git a/...low-core/streamflow-engine/src/main/java/streamflow/engine/framework/FrameworkModule.java b/...low-core/streamflow-engine/src/main/java/streamflow/engine/framework/FrameworkModule.java
@@ -39,15 +39,12 @@ public class FrameworkModule extends AbstractModule {
 
     private final TopologyComponent component;
 
-    private final boolean isCluster;
-
     private final StreamflowConfig streamflowConfig;
 
     public FrameworkModule(Topology topology, TopologyComponent component,
-            boolean isCluster, StreamflowConfig streamflowConfig) {
+            StreamflowConfig streamflowConfig) {
         this.topology = topology;
         this.component = component;
-        this.isCluster = isCluster;
         this.streamflowConfig = streamflowConfig;
     }
 
@@ -92,11 +89,6 @@ protected void configure() {
                 bindConstant().annotatedWith(
                         Names.named("streamflow.cluster.displayName")).to(cluster.getDisplayName());
             }
-
-            if (cluster.getJmsURI() != null) {
-                bindConstant().annotatedWith(
-                        Names.named("streamflow.cluster.jmsUri")).to(cluster.getJmsURI());
-            }
         }
     }
 
@@ -113,7 +105,7 @@ public org.slf4j.Logger provideLogger() {
         // TODO: NEED TO ENSURE THE PROJECT ID IS PASSED TO THE FRAMEWORK MODULE
 
         String logPath = streamflowConfig.getLogger().getBaseDir() 
-                + File.separator + topology.getId() + ".log";
+                + File.separator + "topology-" + topology.getId() + ".log";
 
         FileAppender<ILoggingEvent> fileAppender = new FileAppender<ILoggingEvent>();
         fileAppender.setName("FILE");

diff --git a/streamflow-core/streamflow-engine/src/main/java/streamflow/engine/wrapper/BaseWrapper.java b/streamflow-core/streamflow-engine/src/main/java/streamflow/engine/wrapper/BaseWrapper.java
@@ -71,7 +71,7 @@ protected T getDelegate() throws FrameworkException {
     private void injectModules() throws FrameworkException {
         // Create the new FrameworkModule to inject proxy and property information
         FrameworkModule frameworkModule = new FrameworkModule(
-                topology, component, isCluster, configuration);
+                topology, component, configuration);
 
         // Create the resource module which will inject resource properties
         ResourceModule resourceModule = new ResourceModule(

diff --git a/streamflow-core/streamflow-model/src/main/java/streamflow/model/TopologyLogCriteria.java b/streamflow-core/streamflow-model/src/main/java/streamflow/model/TopologyLogCriteria.java
@@ -33,6 +33,8 @@ public class TopologyLogCriteria implements Serializable {
 
     private int pageSize = 100;
 
+    private boolean showHistoric = false;
+
     private SortOrder sortOrder = SortOrder.DESC;
 
     public enum SortOrder {
@@ -100,6 +102,14 @@ public void setPageSize(int pageSize) {
         this.pageSize = pageSize;
     }
 
+    public boolean getShowHistoric() {
+        return showHistoric;
+    }
+
+    public void setShowHistoric(boolean showHistoric) {
+        this.showHistoric = showHistoric;
+    }
+
     public SortOrder getSortOrder() {
         return sortOrder;
     }
@@ -118,6 +128,7 @@ public int hashCode() {
         hash = 59 * hash + (this.age != null ? this.age.hashCode() : 0);
         hash = 59 * hash + this.pageNum;
         hash = 59 * hash + this.pageSize;
+        hash = 59 * hash + (this.showHistoric ? 1 : 0);
         hash = 59 * hash + (this.sortOrder != null ? this.sortOrder.hashCode() : 0);
         return hash;
     }
@@ -152,6 +163,9 @@ public boolean equals(Object obj) {
         if (this.pageSize != other.pageSize) {
             return false;
         }
+        if (this.showHistoric != other.showHistoric) {
+            return false;
+        }
         if (this.sortOrder != other.sortOrder) {
             return false;
         }
@@ -163,6 +177,6 @@ public String toString() {
         return "TopologyLogCriteria{" + "query=" + query + ", component=" + component 
                 + ", level=" + level + ", category=" + category + ", age=" + age 
                 + ", pageNum=" + pageNum + ", pageSize=" + pageSize 
-                + ", sortOrder=" + sortOrder + '}';
+                + ", showHistoric=" + showHistoric + ", sortOrder=" + sortOrder + '}';
     }
 }
diff --git a/streamflow-core/streamflow-model/src/main/java/streamflow/model/config/LoggerConfig.java b/streamflow-core/streamflow-model/src/main/java/streamflow/model/config/LoggerConfig.java
@@ -27,7 +27,7 @@ public class LoggerConfig implements Serializable {
             + File.separator + ".streamflow" + File.separator + "logs";
 
     private String formatPattern = 
-            "%d{ISO8601,GMT} %p %X{topology} %X{component} %c - %m%n";
+            "%d{ISO8601,GMT} %p %X{topology} %X{project} %X{component} %c - %m%n";
 
     public LoggerConfig() {
     }

diff --git a/streamflow-core/streamflow-server/src/main/webapp/app/topology/topology.tpl.log.cluster.html b/streamflow-core/streamflow-server/src/main/webapp/app/topology/topology.tpl.log.cluster.html
@@ -30,7 +30,7 @@ <h2 class="topology-view-name">
 
 <div class="topology-log-filters clearfix">
     <form class="form-inline well well-sm form-search" role="form">
-        <div class="form-group">
+        <div class="form-group" style="padding-right: 10px;">
             <label for="pageSize">Show: </label>
             <select id="pageSize" class="form-control" ng-model="criteria.pageSize">
                 <option>10</option>
@@ -39,16 +39,23 @@ <h2 class="topology-view-name">
                 <option>100</option>
             </select>
         </div>
-        &nbsp;&nbsp;
-        <div class="form-group">
+        <div class="form-group" style="padding-right: 10px;">
             <label for="sortOrder">Sort By: </label>
             <select id="sortOrder" class="form-control" ng-model="criteria.sortOrder">
                 <option value="DESC">Newest First</option>
                 <option value="ASC">Oldest First</option>
             </select>
         </div>
-        &nbsp;&nbsp;
-        <div class="form-group">
+        <div class="form-group" style="padding-right: 10px;">
+            <div class="checkbox">
+                <label>
+                    <input id="showHistoric" ng-model="criteria.showHistoric" type="checkbox" 
+                        class="form-control"/> Show Historic
+                </label>
+            </div>
+        </div>
+
+        <div class="form-group" style="padding-right: 10px;">
             <label for="query">Query:</label>
             <input id="query" ng-model="criteria.query" type="text" 
                    class="form-control search-query"/>

diff --git a/streamflow-core/streamflow-service/src/main/java/streamflow/service/LogService.java b/streamflow-core/streamflow-service/src/main/java/streamflow/service/LogService.java
@@ -65,7 +65,8 @@ public TopologyLog getTopologyLogLocal(Topology topology, long offset, long limi
 
         try {
             inputReader = new BufferedReader(new FileReader(new File(
-                    streamflowConfig.getLogger().getBaseDir(), topology.getId() + ".log")));
+                    streamflowConfig.getLogger().getBaseDir(), 
+                    "topology-" + topology.getId() + ".log")));
 
             String currentLine;
 
@@ -119,6 +120,9 @@ public TopologyLogPage getTopologyLogCluster(Topology topology, Cluster cluster,
         if (criteria.getCategory() != null) {
             query.must(QueryBuilders.termQuery("category.raw", criteria.getCategory()));
         }
+        if (!criteria.getShowHistoric()) {
+            query.must(QueryBuilders.termQuery("project.raw", topology.getProjectId()));
+        }
 
         // TODO: HANDLE THE AGE CRITERIA
 
@@ -158,19 +162,20 @@ public TopologyLogPage getTopologyLogCluster(Topology topology, Cluster cluster,
 
     public void clearTopologyLog(Topology topology, Cluster cluster) {
         File logFile = new File(streamflowConfig.getLogger().getBaseDir(), 
-                topology.getId() + ".log");
+                "topology-" + topology.getId() + ".log");
 
         try {
             // Delete the local log file from the server
             FileUtils.forceDelete(logFile);
-
         } catch (IOException ex) {
             LOG.error("Error deleting local topology log file: " + logFile.getAbsolutePath());
         }
 
         if (cluster != null) {
             // If the topology was deployed to the cluster also clear the log entries from the server
             if (!cluster.getId().equals(Cluster.LOCAL)) {
+                // Disable delete of log data from elasticsearch for historic purposes
+                /*
                 try {
                     Response searchResponse = Request.Delete(
                             String.format("http://%s:%d/_all/topology/_query?q=topology:%s", 
@@ -183,6 +188,7 @@ public void clearTopologyLog(Topology topology, Cluster cluster) {
                 } catch (IOException ex) {
                     LOG.error("Error deleting cluster topology log data from server: " + ex.getMessage());
                 }
+                */
             }
         }
     }

diff --git a/streamflow-dist/src/main/assembly/common-bin.xml b/streamflow-dist/src/main/assembly/common-bin.xml
@@ -83,6 +83,11 @@
             </includes>
             <filtered>true</filtered>
         </fileSet>
+        <fileSet>
+            <directory>src/main/resources/deploy/docker</directory>
+            <outputDirectory>/deploy/docker</outputDirectory>
+            <filtered>false</filtered>
+        </fileSet>
     </fileSets>
     <files>
         <file>

diff --git a/streamflow-dist/src/main/resources/deploy/docker/storm/fig.yml b/streamflow-dist/src/main/resources/deploy/docker/storm/fig.yml
@@ -0,0 +1,42 @@
+zookeeper:
+  image: wurstmeister/zookeeper
+  ports: 
+    - "49181:2181"
+    - "22"
+nimbus:
+  image: wurstmeister/storm-nimbus
+  ports:
+    - "49773:3773"
+    - "49772:3772"
+    - "49627:6627"
+    - "22"
+  links: 
+    - zookeeper:zk
+supervisor:
+  image: wurstmeister/storm-supervisor
+  ports:
+    - "8000"
+    - "22"
+  links: 
+    - nimbus:nimbus
+    - zookeeper:zk
+  volumes:
+    - /var/log/storm
+ui:
+  image: wurstmeister/storm-ui
+  ports:
+    - "49080:8080"
+    - "22"
+  links: 
+    - nimbus:nimbus
+    - zookeeper:zk
+logstash:
+  image: pblittle/docker-logstash
+  ports: 
+    - "49200:9200"
+    - "49292:9292"
+  volumes_from:
+    - supervisor
+  volumes:
+    - logstash:/opt/logstash/conf.d
+
diff --git a/streamflow-dist/src/main/resources/deploy/docker/storm/logstash/logstash.conf b/streamflow-dist/src/main/resources/deploy/docker/storm/logstash/logstash.conf
@@ -0,0 +1,44 @@
+input {
+    file {
+        type => "topology"
+        path => ["/var/log/storm/topology-*.log"]
+    }
+    file {
+        type => "nimbus"
+        path => ["/var/log/storm/nimbus.log"]
+    }
+    file {
+        type => "supervisor"
+        path => ["/var/log/storm/supervisor.log"]
+    }
+    file {
+        type => "worker"
+        path => ["/var/log/storm/worker-*.log"]
+    }
+}
+
+filter {
+    if [type] == "topology" {
+        multiline {
+            pattern => "^%{TIMESTAMP_ISO8601} "
+            negate => true
+            what => previous
+            enable_flush => true
+        }
+        grok {
+            match => ["message", "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{DATA:topology} %{DATA:project} %{DATA:component} %{DATA:category} - %{GREEDYDATA:text}"]
+        }
+    }
+}
+
+output {
+    stdout {
+        codec => "json_lines"
+    }
+    elasticsearch {
+        embedded => true
+        host => "127.0.0.1"
+        port => "9200"
+        protocol => "http"
+    }
+}