package nl.us2.cloudpelican.stormprocessor;

/**
 * Created by robin on 07/06/15.
 */

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import de.daslaboratorium.machinelearning.classifier.BayesClassifier;
import de.daslaboratorium.machinelearning.classifier.Classifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 *
 * @author robin
 */
public class ErrorClassifierBolt extends BaseRichBolt {

    OutputCollector _collector;
    private Settings settings;
    private HashMap<String, Classifier<String, String>> classifiers;
    private HashMap<String, Long> samplesTrained;
    private String[] errorWords;
    public static final String CLASSIFY_ERROR = "error";
    public static final String CLASSIFY_REGULAR = "regular";
    public static final long MIN_TRAIN_COUNT = 100L;

    private static final Logger LOG = LoggerFactory.getLogger(ErrorClassifierBolt.class);

    public ErrorClassifierBolt(Settings settings) {
        super();
        this.settings = settings;
        this.classifiers = new HashMap<String, Classifier<String, String>>();
        this.samplesTrained = new HashMap<String, Long>();
        this.errorWords = ("err;error;fail;failed;failure;timed out;exception;unexpected;not found;unauthorized;not authorized;missing;reject;rejected;drop;dropped;warn;warning;crit;critical;fatal;emerg;emergency;alert;404").split(";");
    }

    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        _collector = collector;
    }

    public void execute(Tuple tuple) {
        String filterId = tuple.getStringByField("filter_id");
        String msg = tuple.getStringByField("msg");
        String msgLower = msg.toLowerCase();

        // Get classifier
        if (!classifiers.containsKey(filterId)) {
            classifiers.put(filterId, new BayesClassifier<String, String>());
            samplesTrained.put(filterId, 0L);
        }

        // Word in blacklist?
        List<String> msgTokens = Arrays.asList(msg.split("\\s+"));
        boolean errorWordMatch = false;
        for (String errorWord : errorWords) {
            if (msgLower.contains(errorWord)) {
                errorWordMatch = true;
                break;
            }
        }

        // Train classifier
        if (errorWordMatch) {
            // Likely error
            LOG.debug("Train likely error: " + msg);
            classifiers.get(filterId).learn(CLASSIFY_ERROR, msgTokens);
        } else {
            // Likely regular message
            LOG.debug("Train likely regular: " + msg);
            classifiers.get(filterId).learn(CLASSIFY_REGULAR, msgTokens);
        }

        // Increment samples trained
        long trainCount = samplesTrained.get(filterId);
        samplesTrained.put(filterId, trainCount + 1L);

        // Match?
        if (trainCount >= MIN_TRAIN_COUNT && classifiers.get(filterId).classify(msgTokens).getCategory().equals(CLASSIFY_ERROR)) {
            LOG.debug("Classified as error: " + msg);
            _collector.emit("error_stats", new Values(filterId, MetricsEnum.ERRROR.getMask(), 1L)); // Counters
        }
        _collector.ack(tuple);
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("error_stats", new Fields("filter_id", "metric", "increment"));
    }
}
@@ -592,85 +592,9 @@
 til.
-Arrays;%0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map
+*
 ;%0A%0A/
@@ -892,16 +892,43 @@
 rWords;%0A
+    private Random random;%0A
     publ
@@ -1782,24 +1782,56 @@
  collector;%0A
+        random  = new Random();%0A
     %7D%0A%0A    p
@@ -2251,94 +2251,363 @@
  // 
-Word in blacklist?%0A        List%3CString%3E msgTokens = Arrays.asList(msg.split(%22%5C%5Cs+%22));%0A
+Current train count%0A        long trainCount = samplesTrained.get(filterId);%0A%0A        // Tokenize message%0A        List%3CString%3E msgTokens = Arrays.asList(msg.split(%22%5C%5Cs+%22));%0A%0A        // Train classifier (all samples for the first 10K, afterwards 1 out of X%0A        if (trainCount %3C 10000L %7C%7C random.nextInt(10) == 1) %7B%0A            // Word in blacklist?%0A    
     
@@ -2642,16 +2642,20 @@
  false;%0A
+    
         
@@ -2688,24 +2688,28 @@
 rorWords) %7B%0A
+    
             
@@ -2760,16 +2760,20 @@
         
+    
 errorWor
@@ -2803,16 +2803,20 @@
         
+    
 break;%0A 
@@ -2818,32 +2818,36 @@
 ak;%0A            
+    
 %7D%0A        %7D%0A%0A   
@@ -2832,35 +2832,43 @@
       %7D%0A        
+ 
-%7D%0A%0A
+   %7D%0A%0A    
         // Train
@@ -2854,37 +2854,38 @@
 %0A            // 
-Train
+Update
  classifier%0A    
@@ -2871,33 +2871,38 @@
 pdate classifier
-%0A
+s%0A    
         if (erro
@@ -2919,32 +2919,36 @@
 ) %7B%0A            
+    
 // Likely error%0A
@@ -2951,32 +2951,36 @@
 ror%0A            
+  
+  
 LOG.debug(%22Train
@@ -2996,32 +2996,36 @@
 rror: %22 + msg);%0A
+    
             clas
@@ -3088,16 +3088,20 @@
         
+    
 %7D else %7B
@@ -3105,32 +3105,36 @@
 e %7B%0A            
+   
+ 
 // Likely regula
@@ -3135,32 +3135,36 @@
 regular message%0A
+    
             LOG.
@@ -3206,32 +3206,36 @@
 g);%0A            
+  
+  
 classifiers.get(
@@ -3272,32 +3272,36 @@
 AR, msgTokens);%0A
+    
         %7D%0A%0A     
@@ -3303,16 +3303,20 @@
         
+    
 // Incre
@@ -3344,60 +3344,8 @@
     
-    long trainCount = samplesTrained.get(filterId);%0A
     
@@ -3394,16 +3394,26 @@
 t + 1L);
+%0A        %7D
 %0A%0A      

Optimize training of classifiers