source: sample/hadoop-0.16/tw/org/nchc/code/SnortBase.java @ 47

/**
 * Program: LogParserGo.java
 * Editor: Waue Chen 
 * From :  NCHC. Taiwn
 * Last Update Date: 07/02/2008
 */
/**
 * Purpose : 
 *  This program will parse your apache log and store it into Hbase.
 * 
 * HowToUse : 
 *  Make sure two thing :
 *  1. Upload apache logs ( /var/log/apache2/access.log* ) to \ 
 *    hdfs (default: /user/waue/apache-log) \
 *   $ bin/hadoop dfs -put /var/log/apache2/ apache-log
 *  2. parameter "dir" in main contains the logs.
 *  3. you should filter the exception contents manually, \ 
 *    ex:  ::1 - - [29/Jun/2008:07:35:15 +0800] "GET / HTTP/1.0" 200 729 "...
 *  
 * Check Result:
 *  Go to hbase console, type : 
 *    hql > select * from apache-log;

 +-------------------------+-------------------------+-------------------------+
 | Row                     | Column                  | Cell                    |
 +-------------------------+-------------------------+-------------------------+
 | 118.170.101.250         | http:agent              | Mozilla/4.0 (compatible;|
 |                         |                         |  MSIE 4.01; Windows 95) |
 +-------------------------+-------------------------+-------------------------+
 | 118.170.101.250         | http:bytesize           | 318                     |
 +-------------------------+-------------------------+-------------------------+
 ..........(skip)........
 +-------------------------+-------------------------+-------------------------+
 | 87.65.93.58             | http:method             | OPTIONS                 |
 +-------------------------+-------------------------+-------------------------+
 | 87.65.93.58             | http:protocol           | HTTP/1.1                |
 +-------------------------+-------------------------+-------------------------+
 | 87.65.93.58             | referrer:-              | *                       |
 +-------------------------+-------------------------+-------------------------+
 | 87.65.93.58             | url:*                   | -                       |
 +-------------------------+-------------------------+-------------------------+
 31 row(s) in set. (0.58 sec)



 */
package tw.org.nchc.code;

import java.io.IOException;
import java.text.ParsePosition;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseAdmin;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTable;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

class Log {
  String gid, sid, version;

  String alert_name, class_type, priority;

  String source, destination, type;

  // String ttl, tos, id, iplen, dgmlen;
  
  String srcport, dstport,tmp;
  public Log(String data) {

    String[] arr = data.split(";");
    this.gid = arr[0];
    this.sid = arr[1];
    this.version = arr[2];
    this.alert_name = arr[3];
    this.class_type = arr[4];
    this.priority = arr[5];
    this.timestamp = getTime(arr[7] + "/" + arr[6] + ":" + arr[8] + ":"
        + arr[9] + ":" + arr[10]);
    this.source = getIP(arr[11]);
    this.srcport = this.tmp;
    this.destination = getIP(arr[12]);
    this.dstport = this.tmp;
    this.type = arr[13];
    // this.ttl = arr[14];
    // this.tos = arr[15];
    // this.id = arr[16];
    // this.iplen = arr[17];
    // this.dgmlen = arr[18];
    
  }
  long timestamp;


  String getIP(String str){
    String res;
    int n = str.indexOf(":");
    if (n == -1) {
      res = str;
      this.tmp = "0";
    } else {
      String[] vec = str.split(":");
      res = vec[0];
      this.tmp = vec[1];
    }
    return res;
  }

  long getTime(String str) {
    SimpleDateFormat sdf = new SimpleDateFormat("dd/MM:HH:mm:ss",
        Locale.TAIWAN);
    Long timestamp = sdf.parse(str, new ParsePosition(0)).getTime();
    return timestamp;
  }
}

// import AccessLogParser
public class SnortBase {
  static HBaseConfiguration conf = new HBaseConfiguration();

  public static final String TABLE = "table.name";

  static String tableName = "flex";

  static HTable table = null;

  public static class MapClass extends MapReduceBase implements
      Mapper<WritableComparable, Text, Text, Writable> {

    @Override
    // MapReduceBase.configure(JobConf job)
    // Default implementation that does nothing.
    public void configure(JobConf job) {
      // String get(String name,String defaultValue)
      // Get the value of the name property. If no such property exists,\
      // then defaultValue is returned.
    }

    public void map(WritableComparable key, Text value,
        OutputCollector<Text, Writable> output, Reporter reporter)
        throws IOException {

      // try {

      Log log = new Log(value.toString());

      // 查看value的值
      // FileWriter out = new FileWriter(new File(
      // "/home/waue/Desktop/snort-result.txt"));
      // out.write(value.toString() + "_time=" + log.timestamp + "\n");
      // out.flush();
      // out.close();

      if (table == null)
        table = new HTable(conf, new Text(tableName));

      long lockId = table.startUpdate(new Text(log.destination));
      table.put(lockId, new Text("id:gid"), log.gid.getBytes());
      table.put(lockId, new Text("id:sid"), log.sid.getBytes());
      table.put(lockId, new Text("id:version"), log.version.getBytes());
      table.put(lockId, new Text("name:name"), log.alert_name.getBytes());
      table
          .put(lockId, new Text("name:class"), log.class_type
              .getBytes());
      table.put(lockId, new Text("id:priority"), log.priority
          .getBytes());
      table.put(lockId, new Text("direction:soure"), log.source.getBytes());
      table.put(lockId, new Text("direction:srcport"), log.srcport.getBytes());
      table.put(lockId, new Text("direction:dstport"), log.dstport.getBytes());
      table.put(lockId, new Text("payload:type"), log.type.getBytes());
      // table.put(lockId, new Text("payload:ttl"), log.ttl.getBytes());
      // table.put(lockId, new Text("payload:tos"), log.tos.getBytes());
      // table.put(lockId, new Text("payload:id"), log.id.getBytes());
      // table.put(lockId, new Text("payload:iplen"), log.iplen
      // .getBytes());
      // table.put(lockId, new Text("payload:dgmlen"), log.dgmlen
      // .getBytes());
      table.commit(lockId, log.timestamp);

      // } catch (Exception e) {
      // e.printStackTrace();
      // }

    }
  }

  // do it to resolve warning : FileSystem.listPaths
  static public Path[] listPaths(FileSystem fsm, Path path)
      throws IOException {
    FileStatus[] fss = fsm.listStatus(path);
    int length = fss.length;
    Path[] pi = new Path[length];
    for (int i = 0; i < length; i++) {
      pi[i] = fss[i].getPath();
    }
    return pi;
  }

  public static void runMapReduce(String tableName, String inpath)
      throws IOException {
    Path tempDir = new Path("/tmp/Mylog/");
    Path InputPath = new Path(inpath);
    FileSystem fs = FileSystem.get(conf);
    JobConf jobConf = new JobConf(conf, SnortBase.class);
    jobConf.setJobName("Snort Parse");
    jobConf.set(TABLE, tableName);
    // 先省略 自動搜尋目錄的功能
    // Path InputDir = new Path(inpath);
    // Path[] in = listPaths(fs, InputDir);
    // if (fs.isFile(InputDir))
    // {
    // jobConf.setInputPath(InputDir);
    // }
    // else{
    // for (int i = 0; i < in.length; i++){
    // if (fs.isFile(in[i])){
    // jobConf.addInputPath(in[i]);
    // } else
    // {
    // Path[] sub = listPaths(fs, in[i]);
    // for (int j = 0; j < sub.length; j++)
    // {
    // if (fs.isFile(sub[j]))
    // {
    // jobConf.addInputPath(sub[j]);
    // } } } } }

    jobConf.setInputPath(InputPath);
    jobConf.setOutputPath(tempDir);
    jobConf.setMapperClass(MapClass.class);
    JobClient client = new JobClient(jobConf);
    ClusterStatus cluster = client.getClusterStatus();
    jobConf.setNumMapTasks(cluster.getMapTasks());
    jobConf.setNumReduceTasks(0);
    fs.delete(tempDir);
    JobClient.runJob(jobConf);
    fs.delete(tempDir);
    fs.close();
  }

  public static void creatTable(String table) throws IOException {
    HBaseAdmin admin = new HBaseAdmin(conf);
    if (!admin.tableExists(new Text(table))) {
      System.out.println("1. " + table
          + " table creating ... please wait");
      HTableDescriptor tableDesc = new HTableDescriptor(table);
      tableDesc.addFamily(new HColumnDescriptor("id:"));
      tableDesc.addFamily(new HColumnDescriptor("name:"));
      tableDesc.addFamily(new HColumnDescriptor("direction:"));
      tableDesc.addFamily(new HColumnDescriptor("payload:"));
      admin.createTable(tableDesc);
    } else {
      System.out.println("1. " + table + " table already exists.");
    }
    System.out.println("2. access_log files fetching using map/reduce");
  }

  public static void main(String[] args) throws IOException, Exception {

    String path = "/user/waue/snort-log/alert_flex_parsed.txt";

    // 先省略掉 parse完後自動上傳部份
    /*
     * SnortParser sp = new
     * SnortParser("/tmp/alert","/tmp/alert_SnortBase"); sp.parseToLine();
     */
    creatTable(tableName);

    runMapReduce(tableName, path);

  }

}