core/src/main/java/org/apache/hadoop/chukwa/datacollection/sender/RetryListOfCollectors.java

/*
 * 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.sender;


import java.io.*;
import java.nio.charset.Charset;
import java.util.*;

import org.apache.hadoop.conf.Configuration;

/***
 * An iterator returning a list of Collectors to try. This class is
 * nondeterministic, since it puts collectors back on the list after some
 * period.
 * 
 * No node will be polled more than once per maxRetryRateMs milliseconds.
 * hasNext() will continue return true if you have not called it recently.
 * 
 * 
 */
public class RetryListOfCollectors implements Iterator<String>, Cloneable {

  int maxRetryRateMs;
  List<String> collectors;
  long lastLookAtFirstNode;
  int nextCollector = 0;
  private String portNo;
  public static final String RETRY_RATE_OPT = "chukwaAgent.connector.retryRate";

  public RetryListOfCollectors(File collectorFile, Configuration conf)
      throws IOException {
    this(conf);
    try {
      FileInputStream fis = new FileInputStream(collectorFile);
      BufferedReader br = new BufferedReader(new InputStreamReader(fis, Charset.forName("UTF-8")));
      String line, parsedline;
      while ((line = br.readLine()) != null) {
        parsedline = canonicalizeLine(line);
        collectors.add(parsedline);
      }
      
      br.close();
    } catch (FileNotFoundException e) {
      System.err.println("Error in RetryListOfCollectors() opening file"
            + collectorFile.getCanonicalPath() + ", double check that you have"
            + "set the CHUKWA_CONF_DIR environment variable. Also, ensure file"
            + " exists and is in classpath");
      throw e;
    } catch (IOException e) {
      System.err
          .println("I/O error in RetryListOfcollectors instantiation in readLine() from specified collectors file");
      throw e;
    }
    shuffleList();
  }

  private String canonicalizeLine(String line) {
    String parsedline;
    if (!line.contains("://")) {
      // no protocol, assume http
      if (line.matches(".*:\\d+.*")) {
        parsedline = "http://" + line+"/";
      } else {
        parsedline = "http://" + line + ":" + portNo;
      }
    } else {
      if (line.matches(".*:\\d+.*")) {
        parsedline = line;
      } else {
        parsedline = line + ":" + portNo;
      }
    }
    if(!parsedline.matches(".*\\w/.*")) //no resource name
      parsedline = parsedline+"/";
    return parsedline;
  }

  /**
   * This is only used for debugging. Possibly it should sanitize urls the same way the other
   * constructor does.
   * @param collectors is list of collector hostname
   * @param conf is Chukwa configuration
   */
  public RetryListOfCollectors(final List<String> collectors, Configuration conf) {
    this(conf);
    this.collectors.addAll(collectors);
    //we don't shuffle the list here -- this constructor is only used for test purposes
  }
  
  public RetryListOfCollectors(Configuration conf) {
    collectors = new ArrayList<String>();
    portNo = conf.get("chukwaCollector.http.port", "8080");
    maxRetryRateMs = conf.getInt(RETRY_RATE_OPT, 15 * 1000);
    lastLookAtFirstNode = 0;
  }

  // for now, use a simple O(n^2) algorithm.
  // safe, because we only do this once, and on smallish lists
  public void shuffleList() {
    ArrayList<String> newList = new ArrayList<String>();
    Random r = new java.util.Random();
    while (!collectors.isEmpty()) {
      int toRemove = r.nextInt(collectors.size());
      String next = collectors.remove(toRemove);
      newList.add(next);
    }
    collectors = newList;
  }

  public boolean hasNext() {
    return collectors.size() > 0
        && ((nextCollector != 0) || (System.currentTimeMillis()
            - lastLookAtFirstNode > maxRetryRateMs));
  }

  public String next() {
    if (hasNext()) {
      int currCollector = nextCollector;
      nextCollector = (nextCollector + 1) % collectors.size();
      if (currCollector == 0)
        lastLookAtFirstNode = System.currentTimeMillis();
      return collectors.get(currCollector);
    } else
      return null;
  }

  public void add(String collector) {
    collectors.add(collector);
  }

  public void remove() {
    throw new UnsupportedOperationException();
    // FIXME: maybe just remove a collector from our list and then
    // FIXME: make sure next doesn't break (i.e. reset nextCollector if
    // necessary)
  }

  /**
   * 
   * @return total number of collectors in list
   */
  int total() {
    return collectors.size();
  }
  
  public RetryListOfCollectors clone() {
    try {
      RetryListOfCollectors clone = (RetryListOfCollectors) super.clone();
      return clone;
    } catch(CloneNotSupportedException e) {
      return null;
    }
  }

}