core/src/test/java/org/apache/hadoop/chukwa/caffe/TestMemoryUsageDetection.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.caffe;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import junit.framework.TestCase;

import org.apache.hadoop.chukwa.util.ExceptionUtil;


/**
 * (1) Run non-stop terasort and teragen  
 * (2) Collect memory usage metrics from hbase every 5 minutes for 10 hours and write to csv files in /caffe-test/train/data
 * (3) Create images of dimension 1000 * 200 from /caffe-test/train/data/*.csv. 
 *     The files are saved in /caffe-test/train/data/*png
 * (4) Train the image using caffe
 *
 */

public class TestMemoryUsageDetection extends TestCase {

  /**
   * Run non-stop terasort and teragen to force memory leak
   */
  public void setUp() {}

  public void tearDown() {}

  public void testMemoryDetection () {
    String dirName = "/caffe-test/train";
    Thread teraSortThread = createTeraSortThread ();
    ExecutorService executor = Executors.newFixedThreadPool(1);
    Future<?> task = executor.submit(teraSortThread);
    collectNodeManagerMetrics (dirName + "/data");
    task.cancel (true);
    executor.shutdown ();
    caffeTrain (dirName);
  }
  
  private Thread createTeraSortThread () {
    Thread teraSortThread = new Thread(new Runnable() {
      public void run(){
        try {
          String target = new String("/caffe-test/tera/tera.sh");
          Runtime rt = Runtime.getRuntime();
          Process proc = rt.exec(target);
          BufferedReader reader = new BufferedReader(new InputStreamReader(proc.getInputStream()));
          String line = "";                       
          while ((line = reader.readLine())!= null) {
            System.out.println(line + "\n");
          }
        } catch (Exception e) {
          fail(ExceptionUtil.getStackTrace(e));
        }
      }
    });
    return teraSortThread;
  }
  
  
  /**
   * Collect memory usage data every 15 min.
   * Stop the timer after 10 hours
   */
  private void collectNodeManagerMetrics(String dirName) {
    int intervalInMilli = 15 * 60 * 1000;
    long timerDurationTime = 10 * 60 * 60 * 1000;
    String hostname = "";
    try {
      hostname = InetAddress.getLocalHost().getHostName();
      System.out.println (hostname);
    } catch (IOException e) {
      fail(ExceptionUtil.getStackTrace(e));
    }
    MetricsCollector collector = new MetricsCollector (intervalInMilli, hostname, dirName);
    collector.start ();
    try {
      Thread.sleep (timerDurationTime);
    } catch (InterruptedException e) {
    }
    collector.cancel ();
      
    // draw images of size 1000 * 200 from the collected csv files
    try {
      ImageCreator generator = new ImageCreator (dirName);
      generator.drawImages ();
    } catch (Exception e) {
      fail(ExceptionUtil.getStackTrace(e));
    }
  }

  /**
   * Train the images
   */
  private void caffeTrain (String dirName) {
    try {
      String target = new String(dirName + "/train.sh");
      Runtime rt = Runtime.getRuntime();
      Process proc = rt.exec(target);
      //proc.waitFor();
      BufferedReader reader = new BufferedReader(new InputStreamReader(proc.getInputStream()));
      String line = "";                       
      while ((line = reader.readLine())!= null) {
        System.out.println(line + "\n");
      }
    } catch (Exception e) {
      fail(ExceptionUtil.getStackTrace(e));
    }    
  }
}