Browse files

output management for emr

  • Loading branch information...
1 parent 99d6bfd commit 087253be5766e74881b7964a726257f407b340cd @erh erh committed May 25, 2012
Showing with 266 additions and 18 deletions.
  1. +32 −0 buildscripts/emr/IOUtil.java
  2. +70 −8 buildscripts/emr/emr.java
  3. +164 −10 buildscripts/emr/emr.py
View
32 buildscripts/emr/IOUtil.java
@@ -2,6 +2,7 @@
import java.io.*;
import java.net.*;
+import java.util.*;
public class IOUtil {
@@ -68,6 +69,37 @@ public static String readStringFully( InputStream in )
}
+ public static Map<String,Object> readPythonSettings( File file )
+ throws IOException {
+
+ String all = readStringFully( new FileInputStream( file ) );
+
+ Map<String,Object> map = new TreeMap<String,Object>();
+
+ for ( String line : all.split( "\n" ) ) {
+ line = line.trim();
+ if ( line.length() == 0 )
+ continue;
+
+ String[] pcs = line.split( "=" );
+ if ( pcs.length != 2 )
+ continue;
+
+ String name = pcs[0].trim();
+ String value = pcs[1].trim();
+
+ if ( value.startsWith( "\"" ) ) {
+ map.put( name , value.substring( 1 , value.length() - 1 ) );
+ }
+ else {
+ map.put( name , Long.parseLong( value ) );
+ }
+
+ }
+
+ return map;
+ }
+
public static String[] runCommand( String cmd , File dir )
throws IOException {
View
78 buildscripts/emr/emr.java
@@ -23,7 +23,9 @@ void copy( MongoSuite c ) {
mongo = c.mongo;
code = c.code;
workingDir = c.workingDir;
+
suite = c.suite;
+
}
void downloadTo( File localDir )
@@ -40,10 +42,12 @@ boolean runTest()
dir.mkdirs();
// download
+ System.out.println( "going to download" );
downloadTo( dir );
-
+
// explode
+ System.out.println( "going to explode" );
IOUtil.runCommand( "tar zxvf " + IOUtil.urlFileName( code ) , dir );
String[] res = IOUtil.runCommand( "tar zxvf " + IOUtil.urlFileName( mongo ) , dir );
for ( String x : res[0].split( "\n" ) ) {
@@ -54,31 +58,69 @@ boolean runTest()
throw new RuntimeException( "rename failed" );
}
+ List<String> cmd = new ArrayList<String>();
+ cmd.add( "/usr/bin/python" );
+ cmd.add( "buildscripts/smoke.py" );
+
+ File log_config = new File( dir , "log_config.py" );
+ System.out.println( "log_config: " + log_config.exists() );
+ if ( log_config.exists() ) {
+
+ java.util.Map<String,Object> properties = IOUtil.readPythonSettings( log_config );
+
+ cmd.add( "--buildlogger-builder" );
+ cmd.add( properties.get( "name" ).toString() );
+
+ cmd.add( "--buildlogger-buildnum" );
+ cmd.add( properties.get( "number" ).toString() );
+
+ cmd.add( "--buildlogger-credentials" );
+ cmd.add( "log_config.py" );
+
+ cmd.add( "--buildlogger-phase" );
+ {
+ int idx = suite.lastIndexOf( "/" );
+ if ( idx < 0 )
+ cmd.add( suite );
+ else
+ cmd.add( suite.substring( 0 , idx ) );
+ }
+
+ }
+
+ cmd.add( suite );
+
+ System.out.println( cmd );
+
+ Process p = Runtime.getRuntime().exec( cmd.toArray( new String[cmd.size()] ) , new String[]{} , dir );
- Process p = Runtime.getRuntime().exec( new String[]{ "/usr/bin/python" , "buildscripts/smoke.py" , suite } , new String[]{} , dir );
List<Thread> threads = new ArrayList<Thread>();
threads.add( new IOUtil.PipingThread( p.getInputStream() , System.out ) );
threads.add( new IOUtil.PipingThread( p.getErrorStream() , System.out ) );
-
- for ( Thread t : threads ) t.start();
+
+ for ( Thread t : threads )
+ t.start();
+
try {
- for ( Thread t : threads ) t.join();
+ for ( Thread t : threads ) {
+ t.join();
+ }
int rc = p.waitFor();
- System.out.println( "\n\nResult: " + rc );
-
return rc == 0;
}
catch ( InterruptedException ie ) {
+ ie.printStackTrace();
throw new RuntimeException( "sad" , ie );
}
-
+
}
public void readFields( DataInput in )
throws IOException {
mongo = in.readUTF();
code = in.readUTF();
workingDir = in.readUTF();
+
suite = in.readUTF();
}
@@ -87,6 +129,7 @@ public void write( final DataOutput out )
out.writeUTF( mongo );
out.writeUTF( code );
out.writeUTF( workingDir );
+
out.writeUTF( suite );
}
@@ -119,6 +162,14 @@ public void map( Text key, MongoSuite value, OutputCollector<Text,IntWritable> o
ip = ip.substring( ip.indexOf( ":" ) + 1 ).trim();
output.collect( new Text( ip ) , new IntWritable(1) );
}
+ catch ( RuntimeException re ) {
+ re.printStackTrace();
+ throw re;
+ }
+ catch ( IOException ioe ) {
+ ioe.printStackTrace();
+ throw ioe;
+ }
finally {
lock.unlock();
}
@@ -254,6 +305,7 @@ public static void main( String[] args ) throws Exception{
String workingDir = "/data/db/emr/";
+
// parse args
int pos = 0;
@@ -291,6 +343,16 @@ public static void main( String[] args ) throws Exception{
System.out.println( "output\t: " + output );
System.out.println( "suites\t: " + suites );
+ if ( false ) {
+ MongoSuite s = new MongoSuite();
+ s.mongo = mongo;
+ s.code = code;
+ s.workingDir = workingDir;
+ s.suite = suites;
+ s.runTest();
+ return;
+ }
+
// main hadoop set
conf.set( "mongo" , mongo );
conf.set( "code" , code );
View
174 buildscripts/emr/emr.py
@@ -13,6 +13,8 @@
import boto
import simples3
+import pymongo
+
def findSettingsSetup():
sys.path.append( "./" )
sys.path.append( "../" )
@@ -52,11 +54,36 @@ def _get_most_recent_tgz( prefix ):
return all[0][0]
+def get_build_info():
+ return ( os.environ.get('MONGO_BUILDER_NAME') , os.environ.get('MONGO_BUILD_NUMBER') )
+
def make_tarball():
m = _get_most_recent_tgz( "mongodb-" )
- c = "test-code-emr.tgz"
- utils.execsys( "tar zcf %s src jstests buildscripts" % c )
+
+ c = "test-code-emr.tgz"
+ tar = "tar zcf %s src jstests buildscripts" % c
+
+ log_config = "log_config.py"
+ if os.path.exists( log_config ):
+ os.unlink( log_config )
+
+ credentials = do_credentials()
+ if credentials:
+
+ builder , buildnum = get_build_info()
+
+ if builder and buildnum:
+
+ file = open( log_config , "wb" )
+ file.write( 'username="%s"\npassword="%s"\n' % credentials )
+ file.write( 'name="%s"\nnumber=%s\n'% ( builder , buildnum ) )
+
+ file.close()
+
+ tar = tar + " " + log_config
+
+ utils.execsys( tar )
return ( m , c )
def _put_ine( bucket , local , remote ):
@@ -111,18 +138,21 @@ def push():
root = "emr/%s/%s" % ( datetime.date.today().strftime("%Y-%m-%d") , os.uname()[0].lower() )
- def make_long_name(local):
+ def make_long_name(local,hash):
pcs = local.rpartition( "." )
- return "%s/%s-%s.%s" % ( root , pcs[0] , _get_status() , pcs[2] )
+ h = _get_status()
+ if hash:
+ h = utils.md5sum( local )
+ return "%s/%s-%s.%s" % ( root , pcs[0] , h , pcs[2] )
- mongo = _put_ine( bucket , mongo , make_long_name( mongo ) )
- test_code = _put_ine( bucket , test_code , make_long_name( test_code ) )
+ mongo = _put_ine( bucket , mongo , make_long_name( mongo , False ) )
+ test_code = _put_ine( bucket , test_code , make_long_name( test_code , True ) )
jar = build_jar()
- jar = _put_ine( bucket , jar , make_long_name( jar ) )
+ jar = _put_ine( bucket , jar , make_long_name( jar , False ) )
setup = "buildscripts/emr/emrnodesetup.sh"
- setup = _put_ine( bucket , setup , make_long_name( setup ) )
+ setup = _put_ine( bucket , setup , make_long_name( setup , True ) )
return mongo , test_code , jar , setup
@@ -178,7 +208,11 @@ def http(path):
syncdir = "build/emrout/" + jobid + "/"
sync_s3( run_s3_path , syncdir )
- print("output in: build/emrout/" + jobid + "/" )
+
+ final_out = "build/emrout/" + jobid + "/"
+
+ print("output in: " + final_out )
+ do_output( final_out )
def sync_s3( remote_dir , local_dir ):
for x in bucket.listdir( remote_dir ):
@@ -203,6 +237,90 @@ def fix_suites( suites ):
fixed.append( name )
return fixed
+def do_credentials():
+ root = "buildbot.tac"
+
+ while len(root) < 40 :
+ if os.path.exists( root ):
+ break
+ root = "../" + root
+
+ if not os.path.exists( root ):
+ return None
+
+ credentials = {}
+ execfile(root, credentials, credentials)
+
+ if "slavename" not in credentials:
+ return None
+
+ if "passwd" not in credentials:
+ return None
+
+ return ( credentials["slavename"] , credentials["passwd"] )
+
+
+def do_output( dir ):
+
+ def go_down( start ):
+ lst = os.listdir(dir)
+ if len(lst) != 1:
+ raise Exception( "sad: " + start )
+ return start + "/" + lst[0]
+
+ while "out" not in os.listdir( dir ):
+ dir = go_down( dir )
+
+ dir = dir + "/out"
+
+ pieces = os.listdir(dir)
+ pieces.sort()
+
+ passed = []
+ failed = []
+ times = {}
+
+ for x in pieces:
+ if not x.startswith( "part" ):
+ continue
+ full = dir + "/" + x
+
+ for line in open( full , "rb" ):
+ if line.find( "-passed" ) >= 0:
+ passed.append( line.partition( "-passed" )[0] )
+ continue
+
+ if line.find( "-failed" ) >= 0:
+ failed.append( line.partition( "-failed" )[0] )
+ continue
+
+ if line.find( "-time-seconds" ) >= 0:
+ p = line.partition( "-time-seconds" )
+ times[p[0]] = p[2].strip()
+ continue
+
+ print( "\t" + line.strip() )
+
+ def print_list(name,lst):
+ print( name )
+ for x in lst:
+ print( "\t%s\t%s" % ( x , times[x] ) )
+
+ print_list( "passed" , passed )
+ print_list( "failed" , failed )
+
+ if do_credentials():
+ builder , buildnum = get_build_info()
+ if builder and buildnum:
+ conn = pymongo.Connection( "bbout1.10gen.cc" )
+ db = conn.buildlogs
+ q = { "builder" : builder , "buildnum" : int(buildnum) }
+ doc = db.builds.find_one( q )
+
+ if doc:
+ print( "\nhttp://buildlogs.mongodb.org/build/%s" % doc["_id"] )
+
+
if __name__ == "__main__":
if len(sys.argv) == 1:
print( "need an arg" )
@@ -220,11 +338,47 @@ def fix_suites( suites ):
elif sys.argv[1] == "fix_suites":
for x in fix_suites( sys.argv[2:] ):
print(x)
+
+ elif sys.argv[1] == "credentials":
+ print( do_credentials() )
+
+ elif sys.argv[1] == "test":
+ m , c = make_tarball()
+ build_jar()
+ cmd = [ "java" , "-cp" , os.environ.get( "CLASSPATH" , "." ) + ":emr.jar" , "emr" ]
+
+ workingDir = "/data/emr/test"
+ cmd.append( "--workingDir" )
+ cmd.append( workingDir )
+ if os.path.exists( workingDir ):
+ shutil.rmtree( workingDir )
+
+ cmd.append( "file://" + os.getcwd() + "/" + m )
+ cmd.append( "file://" + os.getcwd() + "/" + c )
+
+ out = "/tmp/emrresults"
+ cmd.append( out )
+ if os.path.exists( out ):
+ shutil.rmtree( out )
+
+ cmd.append( "jstests/basic1.js" )
+
+ subprocess.call( cmd )
+ for x in os.listdir( out ):
+ if x.startswith( "." ):
+ continue
+ print( x )
+ for z in open( out + "/" + x ):
+ print( "\t" + z.strip() )
+
+ elif sys.argv[1] == "output":
+ do_output( sys.argv[2] )
+
elif sys.argv[1] == "full":
things = push()
run_tests( things , sys.argv[2:] )
-
+
else:
things = push()
run_tests( things , sys.argv[1:] )

0 comments on commit 087253b

Please sign in to comment.