Exec module changes/fixes.

- Give Dir::monitor() a param for the polling interval, so different
  dirs can be monitored at different frequencies.

- Fix race in Exec::run() when reading extra output files produced by
  a process -- it was possible for Exec::run() to return before all
  extra output files had been fully read.

- Add test cases.

scripts/base/utils/active-http.bro

@@ -90,7 +90,10 @@ function request(req: Request): ActiveHTTP::Response
 		{
 		# If there is no response line then nothing else will work either.
 		if ( ! (result?$files && headersfile in result$files) )
+			{
 			Reporter::error(fmt("There was a failure when requesting \"%s\" with ActiveHTTP.", req$url));
+			return resp;
+			}
 
 		local headers = result$files[headersfile];
 		for ( i in headers )

scripts/base/utils/dir.bro

@@ -5,6 +5,10 @@
 module Dir;
 
 export {
+	## The default interval this module checks for files in directories when
+	## using the :bro:see:`Dir::monitor` function.
+	const polling_interval = 30sec &redef;
+
 	## Register a directory to monitor with a callback that is called 
 	## every time a previously unseen file is seen.  If a file is deleted
 	## and seen to be gone, the file is available for being seen again in 
@@ -14,14 +18,15 @@ export {
 	##
 	## callback: Callback that gets executed with each file name 
 	##           that is found.  Filenames are provided with the full path.
-	global monitor: function(dir: string, callback: function(fname: string));
+	##
-
+	## poll_interval: An interval at which to check for new files.
-	## The interval this module checks for files in directories when using 
+	global monitor: function(dir: string, callback: function(fname: string),
-	## the :bro:see:`Dir::monitor` function.
+	                         poll_interval: interval &default=polling_interval);
-	const polling_interval = 30sec &redef;
 }
 
-event Dir::monitor_ev(dir: string, last_files: set[string], callback: function(fname: string))
+event Dir::monitor_ev(dir: string, last_files: set[string],
+                      callback: function(fname: string),
+                      poll_interval: interval)
 	{
 	when ( local result = Exec::run([$cmd=fmt("ls -i \"%s/\"", str_shell_escape(dir))]) )
 		{
@@ -32,7 +37,11 @@ event Dir::monitor_ev(dir: string, last_files: set[string], callback: function(f
 			}
 
 		local current_files: set[string] = set();
-		local files = result$stdout;
+		local files: vector of string = vector();
+
+		if ( result?$stdout )
+			files = result$stdout;
+
 		for ( i in files )
 			{
 			local parts = split1(files[i], / /);
@@ -40,13 +49,18 @@ event Dir::monitor_ev(dir: string, last_files: set[string], callback: function(f
 				callback(build_path_compressed(dir, parts[2]));
 			add current_files[parts[1]];
 			}
-		schedule polling_interval { Dir::monitor_ev(dir, current_files, callback) };
-		}
-	}
 
-function monitor(dir: string, callback: function(fname: string))
+		schedule poll_interval
 			{
-	event Dir::monitor_ev(dir, set(), callback);
+			Dir::monitor_ev(dir, current_files, callback, poll_interval)
+			};
+		}
+	}
+
+function monitor(dir: string, callback: function(fname: string),
+                 poll_interval: interval &default=polling_interval)
+	{
+	event Dir::monitor_ev(dir, set(), callback, poll_interval);
 	}
 
 

scripts/base/utils/exec.bro

@@ -14,6 +14,8 @@ export {
 		## If additional files are required to be read in as part of the output
 		## of the command they can be defined here.
 		read_files:  set[string] &optional;
+		# The unique id for tracking executors.
+		uid: string &default=unique_id("");
 	};
 
 	type Result: record {
@@ -44,14 +46,11 @@ export {
 	const tmp_dir = "/tmp" &redef;
 }
 
-redef record Command += {
+# Indexed by command uid.
-	# The unique id for tracking executors.
+global results: table[string] of Result;
-	uid: string &optional;
+global pending_commands: set[string];
-};
+global pending_files: table[string] of set[string];
 
-global results: table[string] of Result = table();
-global finished_commands: set[string];
-global currently_tracked_files: set[string] = set();
 type OneLine: record {
 	s: string;
 	is_stderr: bool;
@@ -96,22 +95,46 @@ event Exec::file_line(description: Input::EventDescription, tpe: Input::Event, s
 		result$files[track_file][|result$files[track_file]|] = s;
 	}
 
+event Input::end_of_data(name: string, source:string)
+	{
+	local parts = split1(name, /_/);
+	name = parts[1];
+
+	if ( name !in pending_commands || |parts| < 2 )
+		return;
+
+	local track_file = parts[2];
+
+	Input::remove(name);
+
+	if ( name !in pending_files )
+		delete pending_commands[name];
+	else
+		{
+		delete pending_files[name][track_file];
+		if ( |pending_files[name]| == 0 )
+			delete pending_commands[name];
+		system(fmt("rm \"%s\"", str_shell_escape(track_file)));
+		}
+	}
+
 event InputRaw::process_finished(name: string, source:string, exit_code:count, signal_exit:bool)
 	{
+	if ( name !in pending_commands )
+		return;
+
+	Input::remove(name);
 	results[name]$exit_code = exit_code;
 	results[name]$signal_exit = signal_exit;
 
-	Input::remove(name);
+	if ( name !in pending_files || |pending_files[name]| == 0 )
-	# Indicate to the "when" async watcher that this command is done.
+		# No extra files to read, command is done.
-	add finished_commands[name];
+		delete pending_commands[name];
-	}
+	else
-
+		for ( read_file in pending_files[name] )
-event Exec::start_watching_file(uid: string, read_file: string)
-	{
 			Input::add_event([$source=fmt("%s", read_file),
-	                  $name=fmt("%s_%s", uid, read_file),
+			                  $name=fmt("%s_%s", name, read_file),
 			                  $reader=Input::READER_RAW,
-	                  $mode=Input::STREAM,
 			                  $want_record=F,
 			                  $fields=FileLine,
 			                  $ev=Exec::file_line]);
@@ -119,16 +142,16 @@ event Exec::start_watching_file(uid: string, read_file: string)
 
 function run(cmd: Command): Result
 	{
-	cmd$uid = unique_id("");
+	add pending_commands[cmd$uid];
 	results[cmd$uid] = [];
 
 	if ( cmd?$read_files )
 		{
 		for ( read_file in cmd$read_files )
 			{
-			add currently_tracked_files[read_file];
+			if ( cmd$uid !in pending_files )
-			system(fmt("touch \"%s\" 2>/dev/null", str_shell_escape(read_file)));
+				pending_files[cmd$uid] = set();
-			schedule 1msec { Exec::start_watching_file(cmd$uid, read_file) };
+			add pending_files[cmd$uid][read_file];
 			}
 		}
 
@@ -144,9 +167,8 @@ function run(cmd: Command): Result
 	                  $want_record=F,
 	                  $config=config_strings]);
 
-	return when ( cmd$uid in finished_commands )
+	return when ( cmd$uid !in pending_commands )
 		{
-		delete finished_commands[cmd$uid];
 		local result = results[cmd$uid];
 		delete results[cmd$uid];
 		return result;
@@ -155,9 +177,8 @@ function run(cmd: Command): Result
 
 event bro_done()
 	{
-	# We are punting here and just deleting any files that haven't been processed yet.
+	# We are punting here and just deleting any unprocessed files.
-	for ( fname in currently_tracked_files )
+	for ( uid in pending_files )
-		{
+		for ( fname in pending_files[uid] )
 			system(fmt("rm \"%s\"", str_shell_escape(fname)));
 	}
-	}

testing/btest/Baseline/scripts.base.utils.active-http/bro..stdout

@@ -0,0 +1,5 @@
+[code=200, msg=OK^M, body=It works!, headers={
+[Server] =  1.0,
+[Content-type] =  text/plain,
+[Date] =  July 22, 2013
+}]

testing/btest/Baseline/scripts.base.utils.dir/bro..stdout

@@ -0,0 +1,10 @@
+new_file1, ../testdir/bye
+new_file1, ../testdir/hi
+new_file1, ../testdir/howsitgoing
+new_file2, ../testdir/bye
+new_file2, ../testdir/hi
+new_file2, ../testdir/howsitgoing
+new_file1, ../testdir/bye
+new_file1, ../testdir/newone
+new_file2, ../testdir/bye
+new_file2, ../testdir/newone

testing/btest/Baseline/scripts.base.utils.exec/bro..stdout

@@ -0,0 +1,7 @@
+test1, [exit_code=0, signal_exit=F, stdout=[done, exit, stop], stderr=<uninitialized>, files={
+[out1] = [insert text here, and here],
+[out2] = [insert more text here, and there]
+}]
+test2, [exit_code=1, signal_exit=F, stdout=[here's something on stdout, some more stdout, last stdout], stderr=[and some stderr, more stderr, last stderr], files=<uninitialized>]
+test3, [exit_code=9, signal_exit=F, stdout=[FML], stderr=<uninitialized>, files=<uninitialized>]
+test4, [exit_code=0, signal_exit=F, stdout=[hibye], stderr=<uninitialized>, files=<uninitialized>]

testing/btest/scripts/base/utils/active-http.test

@@ -0,0 +1,25 @@
+# @TEST-EXEC: btest-bg-run httpd python $SCRIPTS/httpd.py --max 1
+# @TEST-EXEC: sleep 3
+# @TEST-EXEC: btest-bg-run bro bro -b %INPUT
+# @TEST-EXEC: btest-bg-wait 15
+# @TEST-EXEC: btest-diff bro/.stdout
+
+@load base/utils/active-http
+
+redef exit_only_after_terminate = T;
+
+event bro_init()
+	{
+	local req = ActiveHTTP::Request($url="localhost:32123");
+
+	when ( local resp = ActiveHTTP::request(req) )
+		{
+		print resp;
+		terminate();
+		}
+	timeout 1min
+		{
+		print "HTTP request timeout";
+		terminate();
+		}
+	}

testing/btest/scripts/base/utils/dir.test

@@ -0,0 +1,58 @@
+# @TEST-EXEC: btest-bg-run bro bro -b ../dirtest.bro
+# @TEST-EXEC: btest-bg-wait 10
+# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff bro/.stdout
+
+@TEST-START-FILE dirtest.bro
+
+@load base/utils/dir
+
+redef exit_only_after_terminate = T;
+
+global c: count = 0;
+
+function check_terminate_condition()
+	{
+	c += 1;
+
+	if ( c == 10 )
+		terminate();
+	}
+
+function new_file1(fname: string)
+	{
+	print "new_file1", fname;
+	check_terminate_condition();
+	}
+
+function new_file2(fname: string)
+	{
+	print "new_file2", fname;
+	check_terminate_condition();
+	}
+
+event change_things()
+	{
+	system("touch ../testdir/newone");
+	system("rm ../testdir/bye && touch ../testdir/bye");
+	}
+
+event bro_init()
+	{
+	Dir::monitor("../testdir", new_file1, .5sec);
+	Dir::monitor("../testdir", new_file2, 1sec);
+	schedule 1sec { change_things() };
+	}
+
+@TEST-END-FILE
+
+@TEST-START-FILE testdir/hi
+123
+@TEST-END-FILE
+
+@TEST-START-FILE testdir/howsitgoing
+abc
+@TEST-END-FILE
+
+@TEST-START-FILE testdir/bye
+!@#
+@TEST-END-FILE

testing/btest/scripts/base/utils/exec.test

@@ -0,0 +1,74 @@
+# @TEST-EXEC: btest-bg-run bro bro -b ../exectest.bro
+# @TEST-EXEC: btest-bg-wait 10
+# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff bro/.stdout
+
+@TEST-START-FILE exectest.bro
+
+@load base/utils/exec
+
+redef exit_only_after_terminate = T;
+
+global c: count = 0;
+
+function check_exit_condition()
+	{
+	c += 1;
+
+	if ( c == 4 )
+		terminate();
+	}
+
+function test_cmd(label: string, cmd: Exec::Command)
+	{
+	when ( local result = Exec::run(cmd) )
+		{
+		print label, result;
+		check_exit_condition();
+		}
+	}
+
+event bro_init()
+	{
+	test_cmd("test1", [$cmd="bash ../somescript.sh",
+	                   $read_files=set("out1", "out2")]);
+	test_cmd("test2", [$cmd="bash ../nofiles.sh"]);
+	test_cmd("test3", [$cmd="bash ../suicide.sh"]);
+	test_cmd("test4", [$cmd="bash ../stdin.sh", $stdin="hibye"]);
+	}
+
+@TEST-END-FILE
+
+@TEST-START-FILE somescript.sh
+#! /usr/bin/env bash
+echo "insert text here" > out1
+echo "and here" >> out1
+echo "insert more text here" > out2
+echo "and there" >> out2
+echo "done"
+echo "exit"
+echo "stop"
+@TEST-END-FILE
+
+@TEST-START-FILE nofiles.sh
+#! /usr/bin/env bash
+echo "here's something on stdout"
+echo "some more stdout"
+echo "last stdout"
+echo "and some stderr" 1>&2
+echo "more stderr" 1>&2
+echo "last stderr" 1>&2
+exit 1
+@TEST-END-FILE
+
+@TEST-START-FILE suicide.sh
+#! /usr/bin/env bash
+echo "FML"
+kill -9 $$
+echo "nope"
+@TEST-END-FILE
+
+@TEST-START-FILE stdin.sh
+#! /usr/bin/env bash
+read -r line
+echo "$line"
+@TEST-END-FILE

testing/scripts/httpd.py

@@ -0,0 +1,40 @@
+#! /usr/bin/env python
+
+import BaseHTTPServer
+
+class MyRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header("Content-type", "text/plain")
+        self.end_headers()
+        self.wfile.write("It works!")
+
+    def version_string(self):
+        return "1.0"
+
+    def date_time_string(self):
+        return "July 22, 2013"
+
+
+if __name__ == "__main__":
+    from optparse import OptionParser
+    p = OptionParser()
+    p.add_option("-a", "--addr", type="string", default="localhost",
+                 help=("listen on given address (numeric IP or host name), "
+                       "an empty string (the default) means INADDR_ANY"))
+    p.add_option("-p", "--port", type="int", default=32123,
+                 help="listen on given TCP port number")
+    p.add_option("-m", "--max", type="int", default=-1,
+                 help="max number of requests to respond to, -1 means no max")
+    options, args = p.parse_args()
+
+    httpd = BaseHTTPServer.HTTPServer((options.addr, options.port),
+                                      MyRequestHandler)
+    if options.max == -1:
+        httpd.serve_forever()
+    else:
+        served_count = 0
+        while served_count != options.max:
+            httpd.handle_request()
+            served_count += 1