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.controller;
  
  
  import java.io.BufferedReader;
  import java.io.File;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.io.OutputStreamWriter;
  import java.io.PrintWriter;
  import java.net.Socket;
  import java.net.SocketException;
  import java.nio.charset.Charset;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Timer;
  import java.util.TimerTask;
  
  import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
  import org.apache.log4j.Logger;
  
  /**
   * A convenience library for applications to communicate to the
   * {@link ChukwaAgent}. Can be used to register and unregister new
   * {@link Adaptor}s. Also contains functions for applications to use for
   * handling log rations.
   */
  public class ChukwaAgentController {
    static Logger log = Logger.getLogger(ChukwaAgentController.class);
     
    public class AddAdaptorTask extends TimerTask {
  
      String adaptorName;
      String type;
      String params;
      private long offset;
      long numRetries;
      long retryInterval;
  
      AddAdaptorTask(String adaptorName, String type, String params, long offset,
                     long numRetries, long retryInterval) {
        this.adaptorName = adaptorName;
        this.type = type;
        this.params = params;
        this.offset = offset;
        this.numRetries = numRetries;
        this.retryInterval = retryInterval;
      }
  
      @Override
      public void run() {
        try {
          log.info("Trying to resend the add command [" + adaptorName + "]["
              + offset + "][" + params + "] [" + numRetries + "]");
          addByName(null, adaptorName, type, params, offset, numRetries, retryInterval);
        } catch (Exception e) {
          log.warn("Exception in AddAdaptorTask.run", e);
          e.printStackTrace();
        }
      }
    }
  
    // our default adaptors, provided here for convenience
    public static final String CharFileTailUTF8 = "org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.CharFileTailingAdaptorUTF8";
    public static final String CharFileTailUTF8NewLineEscaped = "org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.CharFileTailingAdaptorUTF8NewLineEscaped";
  
    static String DEFAULT_FILE_TAILER = CharFileTailUTF8NewLineEscaped;
    static int DEFAULT_PORT = 9093;
    static String DEFAULT_HOST = "localhost";
    static int numArgs = 0;
  
    class Adaptor {
      public String id;
      final public String className;
      final public String params;
      final public String appType;
      public long offset;
  
      Adaptor(String className, String appType, String params, long offset) {
        this.className = className;
        this.appType = appType;
        this.params = params;
        this.offset = offset;
      }
  
      Adaptor(String id, String className, String appType, String params,
              long offset) {
        this.id = id;
        this.className = className;
        this.appType = appType;
        this.params = params;
        this.offset = offset;
      }
  
      /**
       * Registers this {@link Adaptor} with the agent running at the specified
       * hostname and portno
       * 
       * @return The id of the this {@link Adaptor}, assigned by the agent
       *         upon successful registration
       * @throws IOException if problem bind to agent controller port
       */
      String register() throws IOException {
        Socket s = new Socket(hostname, portno);
        try {
          s.setSoTimeout(60000);
        } catch (SocketException e) {
          log.warn("Error while settin soTimeout to 60000");
          e.printStackTrace();
        }
        PrintWriter bw = new PrintWriter(new OutputStreamWriter(s
            .getOutputStream(), Charset.forName("UTF-8")));
        if(id != null)
          bw.println("ADD " + id + " = " + className + " " + appType + " " + params + " " + offset);
        else
          bw.println("ADD " + className + " " + appType + " " + params + " " + offset);
        bw.flush();
        BufferedReader br = new BufferedReader(new InputStreamReader(s
            .getInputStream(), Charset.forName("UTF-8")));
        String resp = br.readLine();
        if (resp != null) {
          String[] fields = resp.split(" ");
          if (fields[0].equals("OK")) {
              id = fields[fields.length - 1];
          }
        }
        s.close();
        return id;
      }
  
      void unregister() throws IOException {
        Socket s = new Socket(hostname, portno);
        try {
          s.setSoTimeout(60000);
        } catch (SocketException e) {
          log.warn("Error while settin soTimeout to 60000");
        }
        PrintWriter bw = new PrintWriter(new OutputStreamWriter(s
            .getOutputStream(), Charset.forName("UTF-8")));
        bw.println("SHUTDOWN " + id);
        bw.flush();
  
        BufferedReader br = new BufferedReader(new InputStreamReader(s
            .getInputStream(), Charset.forName("UTF-8")));
        String resp = br.readLine();
        if (resp == null || !resp.startsWith("OK")) {
          log.error("adaptor unregister error, id: " + id);
        } else if (resp.startsWith("OK")) {
          String[] respSplit = resp.split(" ");
          String newOffset = respSplit[respSplit.length - 1];
          try {
            offset = Long.parseLong(newOffset);
          } catch (NumberFormatException nfe) {
            log.error("adaptor didn't shutdown gracefully.\n" + nfe);
          }
        }
  
        s.close();
      }
  
      public String toString() {
        String[] namePieces = className.split("\\.");
        String shortName = namePieces[namePieces.length - 1];
        return id + " " + shortName + " " + appType + " " + params + " " + offset;
      }
    }
  
    Map<String, ChukwaAgentController.Adaptor> runningAdaptors = new HashMap<String, Adaptor>();
    Map<String, ChukwaAgentController.Adaptor> runningInstanceAdaptors = new HashMap<String, Adaptor>();
    Map<String, ChukwaAgentController.Adaptor> pausedAdaptors;
    String hostname;
    int portno;
  
    public ChukwaAgentController() {
      portno = DEFAULT_PORT;
      hostname = DEFAULT_HOST;
      pausedAdaptors = new HashMap<String, Adaptor>();
  
      syncWithAgent();
    }
  
    public ChukwaAgentController(String hostname, int portno) {
      this.hostname = hostname;
      this.portno = portno;
      pausedAdaptors = new HashMap<String, Adaptor>();
  
      syncWithAgent();
    }
  
    private boolean syncWithAgent() {
      // set up adaptors by using list here
      try {
        runningAdaptors = list();
        return true;
      } catch (IOException e) {
        System.err.println("Error initializing ChukwaClient with list of "
                + "currently registered adaptors, clearing our local list of adaptors");
        // e.printStackTrace();
        // if we can't connect to the LocalAgent, reset/clear our local view of
        // the Adaptors.
        runningAdaptors = new HashMap<String, ChukwaAgentController.Adaptor>();
        return false;
      }
    }
  
    /**
     * Registers a new adaptor. Makes no guarantee about success. On failure, we
     * print a message to stderr and ignore silently so that an application
     * doesn't crash if it's attempt to register an adaptor fails. This call does
     * not retry a conection. for that use the overloaded version of this which
     * accepts a time interval and number of retries
     * @param adaptorName is adaptor class name
     * @param type is data type
     * @param params is adaptor specific parameters
     * @param offset is starting sequence id
     * 
     * @return the id number of the adaptor, generated by the agent
     */
    public String add(String adaptorName, String type, String params, long offset) {
      return addByName(null, adaptorName, type, params, offset, 20, 15 * 1000);// retry for
                                                                   // five
                                                                   // minutes,
                                                                   // every
                                                                   // fifteen
                                                                   // seconds
    }
  
    /**
     * Registers a new adaptor. Makes no guarantee about success. On failure, to
     * connect to server, will retry <code>numRetries</code> times, every
     * <code>retryInterval</code> milliseconds.
     * @param adaptorID is unique adaptor identifier
     * @param adaptorName is adaptor class name
     * @param type is user defined data type name
     * @param params is adaptor specific configuration
     * @param offset is starting sequence id
     * @param numRetries is number of retries
     * @param retryInterval is time between retries
     * 
     * @return the id number of the adaptor, generated by the agent
     */
    public String addByName(String adaptorID, String adaptorName, String type, String params, long offset,
        long numRetries, long retryInterval) {
      ChukwaAgentController.Adaptor adaptor = new ChukwaAgentController.Adaptor(
          adaptorName, type, params, offset);
      adaptor.id = adaptorID;
      if (numRetries >= 0) {
        try {
          adaptorID = adaptor.register();
  
          if (adaptorID != null) {
            runningAdaptors.put(adaptorID, adaptor);
            runningInstanceAdaptors.put(adaptorID, adaptor);
          } else {
            System.err.println("Failed to successfully add the adaptor in AgentClient, adaptorID returned by add() was negative.");
          }
        } catch (IOException ioe) {
          log.warn("AgentClient failed to contact the agent ("
              + hostname + ":" + portno + ")");
          
          log.warn("Scheduling a agent connection retry for adaptor add() in another "
                  + retryInterval
                  + " milliseconds, "
                  + numRetries
                  + " retries remaining");
  
          Timer addFileTimer = new Timer();
          addFileTimer.schedule(new AddAdaptorTask(adaptorName, type, params,
              offset, numRetries - 1, retryInterval), retryInterval);
        }
      } else {
        System.err.println("Giving up on connecting to the local agent");
      }
      return adaptorID;
    }
  
    public synchronized ChukwaAgentController.Adaptor remove(String adaptorID)
        throws IOException {
      syncWithAgent();
      ChukwaAgentController.Adaptor a = runningAdaptors.remove(adaptorID);
      if ( a != null ) {
        a.unregister();
      }
      return a;
  
    }
  
    public void remove(String className, String appType, String filename)
        throws IOException {
      syncWithAgent();
      // search for FileTail adaptor with string of this file name
      // get its id, tell it to unregister itself with the agent,
      // then remove it from the list of adaptors
      for (Adaptor a : runningAdaptors.values()) {
        if (a.className.equals(className) && a.params.equals(filename)
            && a.appType.equals(appType)) {
          remove(a.id);
        }
      }
    }
  
    public void removeAll() {
      syncWithAgent();
      ArrayList<String> keyset = new ArrayList<String>();
      keyset.addAll( runningAdaptors.keySet());
  
      for (String id : keyset) {
        try {
          remove(id);
        } catch (IOException ioe) {
          System.err.println("Error removing an adaptor in removeAll()");
          ioe.printStackTrace();
        }
        log.info("Successfully removed adaptor " + id);
      }
    }
  
    public void removeInstanceAdaptors() {
      // Remove adaptors created by this instance of chukwa agent controller.
      // Instead of removing using id, this is removed by using the stream name
      // and record type.  This prevents the system to shutdown the wrong
      // adaptor after agent crashes.
      for (Adaptor a : runningInstanceAdaptors.values()) {
        try {
          remove(a.className, a.appType, a.params);
        } catch (IOException ioe) {
          log.warn("Error removing an adaptor in removeInstanceAdaptors()");
          ioe.printStackTrace();
        }
      }
    }
  
    Map<String, ChukwaAgentController.Adaptor> list() throws IOException {
      Socket s = new Socket(hostname, portno);
      try {
        s.setSoTimeout(60000);
      } catch (SocketException e) {
        log.warn("Error while settin soTimeout to 60000");
        e.printStackTrace();
      }
      PrintWriter bw = new PrintWriter(
          new OutputStreamWriter(s.getOutputStream(), Charset.forName("UTF-8")));
  
      bw.println("LIST");
      bw.flush();
      BufferedReader br = new BufferedReader(new InputStreamReader(s
          .getInputStream(), Charset.forName("UTF-8")));
      String ln;
      Map<String, Adaptor> listResult = new HashMap<String, Adaptor>();
      while ((ln = br.readLine()) != null) {
        if (ln.equals("")) {
          break;
        } else {
          String[] parts = ln.split("\\s+");
          if (parts.length >= 4) { // should have id, className appType, params,
                                   // offset
            String id = parts[0].substring(0, parts[0].length() - 1); // chop
                                                                          // off
                                                                          // the
                                                                          // right
                                                                          // -
                                                                          // paren
            long offset = Long.parseLong(parts[parts.length - 1]);
            StringBuilder tmpParams = new StringBuilder();
            tmpParams.append(parts[3]);
            for (int i = 4; i < parts.length - 1; i++) {
              tmpParams.append(" ");
              tmpParams.append(parts[i]);
            }
            listResult.put(id, new Adaptor(id, parts[1], parts[2], tmpParams.toString(),
                offset));
          }
        }
      }
      s.close();
      return listResult;
    }
  
    // ************************************************************************
    // The following functions are convenience functions, defining an easy
    // to use API for application developers to integrate chukwa into their app
    // ************************************************************************
  
    /**
     * Registers a new "LineFileTailUTF8" adaptor and starts it at offset 0.
     * Checks to see if the file is being watched already, if so, won't register
     * another adaptor with the agent. If you have run the tail adaptor on this
     * file before and rotated or emptied the file you should use
     * {@link ChukwaAgentController#pauseFile(String, String)} and
     * {@link ChukwaAgentController#resumeFile(String, String)} which will store
     * the adaptors metadata and re-use them to pick up where it left off.
     * @param appType is user defined name for the data stream
     * 
     * @param filename of the file for the tail adaptor to start monitoring
     * @param numRetries is number of retries
     * @param retryInterval is time between retries
     * @return the id number of the adaptor, generated by the agent
     */
    public String addFile(String appType, String filename, long numRetries,
        long retryInterval) {
      filename = new File(filename).getAbsolutePath();
      // TODO: Mabye we want to check to see if the file exists here?
      // Probably not because they might be talking to an agent on a different
      // machine?
  
      // check to see if this file is being watched already, if yes don't set up
      // another adaptor for it
      boolean isDuplicate = false;
      for (Adaptor a : runningAdaptors.values()) {
        if (a.className.equals(DEFAULT_FILE_TAILER) && a.appType.equals(appType)
            && a.params.endsWith(filename)) {
          isDuplicate = true;
        }
      }
      if (!isDuplicate) {
        return addByName(null, DEFAULT_FILE_TAILER, appType, 0L + " " + filename, 0L,
            numRetries, retryInterval);
      } else {
        log.info("An adaptor for filename \"" + filename
            + "\", type \"" + appType
            + "\", exists already, addFile() command aborted");
        return null;
      }
    }
  
    public String addFile(String appType, String filename) {
      return addFile(appType, filename, 0, 0);
    }
  
    /**
     * Pause all active adaptors of the default file tailing type who are tailing
     * this file This means we actually stop the adaptor and it goes away forever,
     * but we store it state so that we can re-launch a new adaptor with the same
     * state later.
     * 
     * @param appType is application type
     * @param filename is file name suffix pattern
     * @return array of adaptorID numbers which have been created and assigned the
     *         state of the formerly paused adaptors
     * @throws IOException if error pausing adaptors
     */
    public Collection<String> pauseFile(String appType, String filename)
        throws IOException {
      syncWithAgent();
      // store the unique streamid of the file we are pausing.
      // search the list of adaptors for this filename
      // store the current offset for it
      List<String> results = new ArrayList<String>();
      for (Adaptor a : runningAdaptors.values()) {
        if (a.className.equals(DEFAULT_FILE_TAILER) && a.params.endsWith(filename)
            && a.appType.equals(appType)) {
          pausedAdaptors.put(a.id, a); // add it to our list of paused adaptors
          remove(a.id); // tell the agent to remove/unregister it
          results.add(a.id);
        }
      }
      return results;
    }
  
    public boolean isFilePaused(String appType, String filename) {
      for (Adaptor a : pausedAdaptors.values()) {
        if (a.className.equals(DEFAULT_FILE_TAILER) && a.params.endsWith(filename)
            && a.appType.equals(appType)) {
          return true;
        }
      }
      return false;
    }
  
    /**
     * Resume all adaptors for this filename that have been paused
     * 
     * @param appType is application type
     * @param filename filename by which to lookup adaptors which are paused (and
     *        tailing this file)
     * @return an array of the new adaptor ID numbers which have resumed where the
     *         old adaptors left off
     * @throws IOException if unable to resume all adaptors
     */
    public Collection<String> resumeFile(String appType, String filename)
        throws IOException {
      syncWithAgent();
      // search for a record of this paused file
      List<String> results = new ArrayList<String>();
      for (Adaptor a : pausedAdaptors.values()) {
        if (a.className.equals(DEFAULT_FILE_TAILER) && a.params.endsWith(filename)
            && a.appType.equals(appType)) {
          String newID = add(DEFAULT_FILE_TAILER, a.appType, a.offset + " "
              + filename, a.offset);
          pausedAdaptors.remove(a.id);
          a.id = newID;
          results.add(a.id);
        }
      }
      return results;
    }
  
    public void removeFile(String appType, String filename) throws IOException {
      syncWithAgent();
      // search for FileTail adaptor with string of this file name
      // get its id, tell it to unregister itself with the agent,
      // then remove it from the list of adaptors
      for (Adaptor a : runningAdaptors.values()) {
        if (a.className.equals(DEFAULT_FILE_TAILER) && a.params.endsWith(filename)
            && a.appType.equals(appType)) {
          remove(a.id);
        }
      }
    }
  
    // ************************************************************************
    // command line utilities
    // ************************************************************************
  
    public static void main(String[] args) {
      ChukwaAgentController c = getClient(args);
      if (numArgs >= 3 && args[0].toLowerCase().equals("addfile")) {
        doAddFile(c, args[1], args[2]);
      } else if (numArgs >= 3 && args[0].toLowerCase().equals("removefile")) {
        doRemoveFile(c, args[1], args[2]);
      } else if (numArgs >= 1 && args[0].toLowerCase().equals("list")) {
        doList(c);
      } else if (numArgs >= 1 && args[0].equalsIgnoreCase("removeall")) {
        doRemoveAll(c);
      } else {
        System.err.println("usage: ChukwaClient addfile <apptype> <filename> [-h hostname] [-p portnumber]");
        System.err.println("       ChukwaClient removefile adaptorID [-h hostname] [-p portnumber]");
        System.err.println("       ChukwaClient removefile <apptype> <filename> [-h hostname] [-p portnumber]");
        System.err.println("       ChukwaClient list [IP] [port]");
        System.err.println("       ChukwaClient removeAll [IP] [port]");
      }
    }
  
    private static ChukwaAgentController getClient(String[] args) {
      int portno = 9093;
      String hostname = "localhost";
  
      numArgs = args.length;
  
      for (int i = 0; i < args.length; i++) {
        if (args[i].equals("-h") && args.length > i + 1) {
          hostname = args[i + 1];
          log.debug("Setting hostname to: " + hostname);
          numArgs -= 2; // subtract for the flag and value
        } else if (args[i].equals("-p") && args.length > i + 1) {
          portno = Integer.parseInt(args[i + 1]);
          log.debug("Setting portno to: " + portno);
          numArgs -= 2; // subtract for the flat, i.e. -p, and value
        }
      }
      return new ChukwaAgentController(hostname, portno);
    }
  
    private static String doAddFile(ChukwaAgentController c, String appType,
        String params) {
      log.info("Adding adaptor with filename: " + params);
      String adaptorID = c.addFile(appType, params);
      if (adaptorID != null) {
        log.info("Successfully added adaptor, id is:" + adaptorID);
      } else {
        log.error("Agent reported failure to add adaptor.");
      }
      return adaptorID;
    }
  
    private static void doRemoveFile(ChukwaAgentController c, String appType,
        String params) {
      try {
        log.debug("Removing adaptor with filename: " + params);
        c.removeFile(appType, params);
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  
    private static void doList(ChukwaAgentController c) {
      try {
        Iterator<Adaptor> adptrs = c.list().values().iterator();
        while (adptrs.hasNext()) {
          log.debug(adptrs.next().toString());
        }
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  
    private static void doRemoveAll(ChukwaAgentController c) {
      log.info("Removing all adaptors");
      c.removeAll();
    }
  }
  