content: /*
   * 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
   * with the License.  You may obtain a copy of the License at
   *
   *     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.hadoop.chukwa.util;
  
  
  import java.net.URI;
  import java.net.URISyntaxException;
  import java.util.regex.*;
  import java.util.*;
  import java.io.*;
  import org.apache.hadoop.chukwa.*;
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.util.RegexUtil.CheckedPatternSyntaxException;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.FileUtil;
  import org.apache.hadoop.io.SequenceFile;
  import org.apache.hadoop.conf.Configuration;
  
  public class DumpChunks {
  
    
    /**
     * Tries to find chunks matching a given pattern.
     * Takes as input a set of &amp;-delimited patterns, followed
     * by a list of file names.
     * 
     * E.g:  Dump datatype=Iostat&amp;source=/my/log/.* *.done
     * @param args is command line parameters
     * @throws IOException if problem access HDFS
     * @throws URISyntaxException if error parsing HDFS URL
     */
    public static void main(String[] args) throws IOException, URISyntaxException {
      
      if(args.length < 2) {
        System.out.println("usage: Dump [-s] pattern1,pattern2,pattern3... file1 file2 file3...");
        return;
      }
      
      ChukwaConfiguration conf = new ChukwaConfiguration();
  
      dump(args, conf, System.out);
    }
    
    static FileSystem getFS(Configuration conf, String uri) throws IOException, URISyntaxException {
      FileSystem fs;
      if(uri.contains("://")) {
        fs = FileSystem.get(new URI(uri), conf);
      } else {
        String fsName = conf.get("writer.hdfs.filesystem");
        if(fsName == null)
          fs = FileSystem.getLocal(conf);
        else
          fs = FileSystem.get(conf);
      }
      System.err.println("filesystem is " + fs.getUri());
      return fs;
    }
  
    static void dump(String[] args, Configuration conf, PrintStream out) throws IOException, URISyntaxException {
      
      int filterArg = 0;
      boolean summarize = false;
      boolean nosort = false;
      if(args[0].equals("-s")) {
        filterArg++;
        summarize = true;
      } else if(args[0].equals("--nosort")) {
        filterArg++;
        nosort = true;
      }
      
      Filter patterns = null;
      if(args[filterArg].toLowerCase().equals("all"))
        patterns = Filter.ALL;
      else {
        try {
          patterns = new Filter(args[filterArg]);
        } catch (CheckedPatternSyntaxException pse) {
          System.err.println("Error parsing \"tags\" regular expression: " + pse.getMessage());
          return;
        }
      }
  
      System.err.println("Patterns:" + patterns);
      ArrayList<Path> filesToSearch = new ArrayList<Path>();
  
      FileSystem fs = getFS(conf, args[filterArg + 1]);
      for(int i=filterArg + 1; i < args.length; ++i){
        Path[] globbedPaths = FileUtil.stat2Paths(fs.globStatus(new Path(args[i])));
        if(globbedPaths != null)
          for(Path p: globbedPaths)
            filesToSearch.add(p);
      }
      
      System.err.println("expands to " + filesToSearch.size() + " actual files");
  
      DumpChunks dc;
      if(summarize)
        dc = new DumpAndSummarize();
      else if(nosort)
        dc = new DumpNoSort(out);
      else
        dc= new DumpChunks();
      
      try {
        for(Path p: filesToSearch) {
        
          SequenceFile.Reader r = new SequenceFile.Reader(fs, p, conf);
    
          ChukwaArchiveKey key = new ChukwaArchiveKey();
          ChunkImpl chunk = ChunkImpl.getBlankChunk();
          while (r.next(key, chunk)) {
            if(patterns.matches(chunk)) {
              dc.updateMatchCatalog(key.getStreamName(), chunk);
              chunk = ChunkImpl.getBlankChunk();
            }
          }
        }
        
        dc.displayResults(out);
        
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  
    public DumpChunks() {
      matchCatalog = new HashMap<String, SortedMap<Long, ChunkImpl> >();
    }
  
    Map<String, SortedMap<Long, ChunkImpl>> matchCatalog;
    
    protected void displayResults(PrintStream out) throws IOException{
      for(Map.Entry<String,SortedMap<Long, ChunkImpl>> streamE: matchCatalog.entrySet()) {
        String header = streamE.getKey();
        SortedMap<Long, ChunkImpl> stream = streamE.getValue();
        long nextToPrint = 0;
        if(stream.firstKey() > 0)
          System.err.println("---- map starts at "+ stream.firstKey());
        for(Map.Entry<Long, ChunkImpl> e: stream.entrySet()) {
          if(e.getKey() >= nextToPrint) {
            if(e.getKey() > nextToPrint)
              System.err.println("---- printing bytes starting at " + e.getKey());
            
            out.write(e.getValue().getData());
            nextToPrint = e.getValue().getSeqID();
          } else if(e.getValue().getSeqID() < nextToPrint) {
            continue; //data already printed
          } else {
            //tricky case: chunk overlaps with already-printed data, but not completely
            ChunkImpl c = e.getValue();
            long chunkStartPos = e.getKey();
            int numToPrint = (int) (c.getSeqID() - nextToPrint);
            int printStartOffset = (int) ( nextToPrint -  chunkStartPos);
            out.write(c.getData(), printStartOffset, numToPrint);
            nextToPrint = c.getSeqID();
          }
        }
        out.println("\n--------"+header + "--------");
      }
    }
   
    protected void updateMatchCatalog(String streamName,  ChunkImpl chunk) throws IOException {
  
      SortedMap<Long, ChunkImpl> chunksInStream = matchCatalog.get(streamName);
      if(chunksInStream == null ) {
        chunksInStream = new TreeMap<Long, ChunkImpl>();
        matchCatalog.put(streamName, chunksInStream);
      }
      
      long startPos = chunk.getSeqID() - chunk.getLength();
      
      ChunkImpl prevMatch = chunksInStream.get(startPos);
      if(prevMatch == null)
        chunksInStream.put(startPos, chunk);
      else { //pick longest
        if(chunk.getLength() > prevMatch.getLength())
          chunksInStream.put (startPos, chunk);
      }
    }
  
    static class DumpAndSummarize extends DumpChunks {
      Map<String, Integer> matchCounts = new LinkedHashMap<String, Integer>();
      Map<String, Long> byteCounts = new LinkedHashMap<String, Long>();
      
  
      protected void displayResults(PrintStream out) throws IOException{
        for(Map.Entry<String, Integer> s: matchCounts.entrySet()) {
          out.print(s.getKey());
          out.print(" ");
          out.print(s.getValue());
          out.print(" chunks ");
          out.print(byteCounts.get(s.getKey()));
          out.println(" bytes");
        }
          
      }
      
      protected void updateMatchCatalog(String streamName,  ChunkImpl chunk) {
        Integer i = matchCounts.get(streamName);
        if(i != null) {
          matchCounts.put(streamName, i+1);
          Long b = byteCounts.get(streamName);
          byteCounts.put(streamName, b + chunk.getLength());
        } else {
          matchCounts.put(streamName, Integer.valueOf(1));
          byteCounts.put(streamName, Long.valueOf(chunk.getLength()));
        }
      }
      
    }
    
    static class DumpNoSort extends DumpChunks {
      
      PrintStream out; 
      public DumpNoSort(PrintStream out) {
        this.out = out;
      }
      //Do some display
      protected void updateMatchCatalog(String streamName,  ChunkImpl chunk) throws IOException {
        out.write(chunk.getData());
      }
      
      protected void displayResults(PrintStream out) throws IOException{
        //did this in updateMatchCatalog
      }
      
    }
  
  }
  