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.datacollection.collector.servlet;
  
  
  import java.io.DataInputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.PrintStream;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Timer;
  import java.util.TimerTask;
  
  import javax.servlet.ServletConfig;
  import javax.servlet.ServletException;
  import javax.servlet.ServletOutputStream;
  import javax.servlet.http.HttpServlet;
  import javax.servlet.http.HttpServletRequest;
  import javax.servlet.http.HttpServletResponse;
  
  import org.apache.hadoop.chukwa.Chunk;
  import org.apache.hadoop.chukwa.ChunkImpl;
  import org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter;
  import org.apache.hadoop.chukwa.datacollection.writer.SeqFileWriter;
  import org.apache.hadoop.chukwa.datacollection.writer.WriterException;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.io.compress.CompressionCodec;
  import org.apache.hadoop.util.ReflectionUtils;
  import org.apache.log4j.Logger;
  
  @Deprecated
  public class ServletCollector extends HttpServlet {
  
    static final boolean FANCY_DIAGNOSTICS = false;
    public static final String PATH = "chukwa";
    /**
     * If a chunk is committed; then the ack will start with the following string.
     */
    public static final String ACK_PREFIX = "ok: ";
    transient ChukwaWriter writer = null;
  
    private static final long serialVersionUID = 6286162898591407111L;
    transient Logger log = Logger.getLogger(ServletCollector.class);
    
    boolean COMPRESS;
    String CODEC_NAME;
    transient CompressionCodec codec;
  
    public void setWriter(ChukwaWriter w) {
      writer = w;
    }
    
    public ChukwaWriter getWriter() {
      return writer;
    }
  
    long statTime = 0L;
    int numberHTTPConnection = 0;
    int numberchunks = 0;
    long lifetimechunks = 0;
  
    transient Configuration conf;
  
    public ServletCollector(Configuration c) {
      conf = c;
    }
  
    public void init(ServletConfig servletConf) throws ServletException {
  
      log.info("initing servletCollector");
      if (servletConf == null) {
        log.fatal("no servlet config");
        return;
      }
  
      Timer statTimer = new Timer();
      statTimer.schedule(new TimerTask() {
        public void run() {
          log.info("stats:ServletCollector,numberHTTPConnection:"
              + numberHTTPConnection + ",numberchunks:" + numberchunks);
          statTime = System.currentTimeMillis();
          numberHTTPConnection = 0;
          numberchunks = 0;
        }
      }, (1000), (60 * 1000));
  
      if (writer != null) {
        log.info("writer set up statically, no need for Collector.init() to do it");
        return;
      }
  
      try {
        String writerClassName = conf.get("chukwaCollector.writerClass",
            SeqFileWriter.class.getCanonicalName());
        Class<?> writerClass = Class.forName(writerClassName);
        if (writerClass != null
            && ChukwaWriter.class.isAssignableFrom(writerClass))
          writer = (ChukwaWriter) writerClass.newInstance();
      } catch (Exception e) {
        log.warn("failed to use user-chosen writer class, defaulting to SeqFileWriter", e);
      }
  
      COMPRESS = conf.getBoolean("chukwaAgent.output.compress", false);
      if( COMPRESS) {
  	    CODEC_NAME = conf.get( "chukwaAgent.output.compression.type", "org.apache.hadoop.io.compress.DefaultCodec");
  	    Class<?> codecClass = null;
  	    try {
  			codecClass = Class.forName( CODEC_NAME);
  			codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
  			log.info("codec " + CODEC_NAME + " loaded for network compression");
  		} catch (ClassNotFoundException e) {
  			log.warn("failed to create codec " + CODEC_NAME + ". Network compression won't be enabled.", e);
  			COMPRESS = false;
  		}
      }
      
      // We default to here if the pipeline construction failed or didn't happen.
      try {
        if (writer == null) {
          writer =  new SeqFileWriter();
        }
        
        writer.init(conf);
      } catch (Throwable e) {
        log.warn("Exception trying to initialize SeqFileWriter",e);
      }
    }
  
    @Override
    protected void doTrace(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { 
      resp.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED); 
    }
  
    protected void accept(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException {
      numberHTTPConnection++;
      final long currentTime = System.currentTimeMillis();
      try {
  
        log.debug("new post from " + req.getRemoteHost() + " at " + currentTime);
        java.io.InputStream in = req.getInputStream();
  
        ServletOutputStream l_out = resp.getOutputStream();
        
        DataInputStream di = null;
        boolean compressNetwork = COMPRESS;
        if( compressNetwork){
            InputStream cin = codec.createInputStream( in);
            di = new DataInputStream(cin);
        }
        else {
      	  di = new DataInputStream(in);
        }
  
        final int numEvents = di.readInt();
        // log.info("saw " + numEvents+ " in request");
  
        List<Chunk> events = new LinkedList<Chunk>();
        StringBuilder sb = new StringBuilder();
  
        for (int i = 0; i < numEvents; i++) {
          ChunkImpl logEvent = ChunkImpl.read(di);
          events.add(logEvent);
  
        }
  
        int responseStatus = HttpServletResponse.SC_OK;
  
        // write new data to data sync file
        if (writer != null) {
          ChukwaWriter.CommitStatus result = writer.add(events);
  
          // this is where we ACK this connection
  
          if(result == ChukwaWriter.COMMIT_OK) {
            // only count the chunks if result is commit or commit pending
            numberchunks += events.size();
            lifetimechunks += events.size();
  
            for(Chunk receivedChunk: events) {
              sb.append(ACK_PREFIX);
              sb.append(receivedChunk.getData().length);
              sb.append(" bytes ending at offset ");
              sb.append(receivedChunk.getSeqID() - 1).append("\n");
            }
          } else if(result instanceof ChukwaWriter.COMMIT_PENDING) {
  
            // only count the chunks if result is commit or commit pending
            numberchunks += events.size();
            lifetimechunks += events.size();
  
            for(String s: ((ChukwaWriter.COMMIT_PENDING) result).pendingEntries)
              sb.append(s);
          } else if(result == ChukwaWriter.COMMIT_FAIL) {
            sb.append("Commit failed");
            responseStatus = HttpServletResponse.SC_SERVICE_UNAVAILABLE;
          }
  
          l_out.print(sb.toString());
        } else {
          l_out.println("can't write: no writer");
        }
  
        resp.setStatus(responseStatus);
  
      } catch (Throwable e) {
        log.warn("Exception talking to " + req.getRemoteHost() + " at t="
            + currentTime, e);
        throw new ServletException(e);
      }
    }
  
    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {
      accept(req, resp);
    }
  
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {
  
  
      log.info("new GET from " + req.getRemoteHost() + " at " + System.currentTimeMillis());
      PrintStream out = new PrintStream(resp.getOutputStream(), true, "UTF-8");
      resp.setStatus(200);
  
      String pingAtt = req.getParameter("ping");
      if (pingAtt != null) {
        out.println("Date:" + statTime);
        out.println("Now:" + System.currentTimeMillis());
        out.println("numberHTTPConnection in time window:"
            + numberHTTPConnection);
        out.println("numberchunks in time window:" + numberchunks);
        out.println("lifetimechunks:" + lifetimechunks);
      } else {
        out.println("<html><body><h2>Chukwa servlet running</h2>");
        out.println("</body></html>");
      }
  
    }
  
    @Override
    public String getServletInfo() {
      return "Chukwa Servlet Collector";
    }
  
    @Override
    public void destroy() {
      try {
        writer.close();
      } catch (WriterException e) {
        log.warn("Exception during close", e);
        e.printStackTrace();
      }
      super.destroy();
    }
  }
  