sumstat/non-cluster: Move last epoch processing to zeek_done()

@Sheco reported that standalone epoch processing may exclude scheduled
events when the final sumstat epoch runs before. For example, this easily
happens when attempting to do sumstat observations within connection_state_remove().

Delay final epoch processing to zeek_done() instead.

This doesn't deal with the clustered version - this would need something
more elaborate and potentially a mechanism to delay the shutdown of
other cluster nodes until/after sumstat processing completed.

scripts/base/frameworks/sumstats/non-cluster.zeek

@@ -27,44 +27,60 @@ event SumStats::process_epoch_result(ss: SumStat, now: time, data: ResultTable)
 		ss$epoch_finished(now);
 	}
 
-event SumStats::finish_epoch(ss: SumStat)
+
+function do_finish_epoch(ss: SumStat)
 	{
-	if ( ss$name in result_store )
+	if ( ss$name !in result_store || ! ss?$epoch_result )
+		return;
+
+	local data = result_store[ss$name];
+	local now = network_time();
+	if ( zeek_is_terminating() )
 		{
-		if ( ss?$epoch_result )
+		for ( key, val in data )
-			{
+			ss$epoch_result(now, key, val);
-			local data = result_store[ss$name];
-			local now = network_time();
-			if ( zeek_is_terminating() )
-				{
-				for ( key, val in data )
-					ss$epoch_result(now, key, val);
 
-				if ( ss?$epoch_finished )
+		if ( ss?$epoch_finished )
-					ss$epoch_finished(now);
+			ss$epoch_finished(now);
-				}
-			else
-				{
-				if ( |data| > 0 )
-					event SumStats::process_epoch_result(ss, now, copy(data));
-				else
-					{
-					if ( ss?$epoch_finished )
-						ss$epoch_finished(now);
-					}
-				}
-			}
-
-		# We can reset here because we know that the reference
-		# to the data will be maintained by the process_epoch_result
-		# event.
-		reset(ss);
 		}
+	else
+		{
+		if ( |data| > 0 )
+			event SumStats::process_epoch_result(ss, now, copy(data));
+		else
+			{
+			if ( ss?$epoch_finished )
+				ss$epoch_finished(now);
+			}
+		}
+	# We can reset here because we know that the reference to the
+	# data will be maintained by the process_epoch_result event.
+	reset(ss);
 
 	if ( ss$epoch != 0secs )
 		schedule ss$epoch { SumStats::finish_epoch(ss) };
 	}
 
+event SumStats::finish_epoch(ss: SumStat)
+	{
+	if ( zeek_is_terminating() )
+		return;  # runs during zeek_done() instead
+
+	do_finish_epoch(ss);
+	}
+
+# Run non-manual SumStats entries as late as possible, but a bit
+# earlier than a user's zeek_done() handler in case they end up
+# doing something curious in zeek_done().
+event zeek_done() &priority=10
+	{
+	for ( name, ss in stats_store )
+		{
+		if ( ss$epoch != 0sec )  # skip SumStats with manual epochs.
+			do_finish_epoch(ss);
+		}
+	}
+
 function data_added(ss: SumStat, key: Key, result: Result)
 	{
 	if ( check_thresholds(ss, key, result, 1.0) )

testing/btest/Baseline/scripts.base.frameworks.sumstats.final-epoch/.stdout

@@ -0,0 +1,60 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+zeek_init
+epoch finished, 1300475167.096535
+net_done, 1300475173.475401
+connection_state_remove, CHhAvVGS1DHFjwGM9
+connection_state_remove, ClEkJM2Vm5giqnMf4h
+connection_state_remove, C4J4Th3PJpwUYZZ6gc
+connection_state_remove, CtPZjS20MLrsMUOJi2
+connection_state_remove, CUM0KZ3MLUfNB0cl11
+connection_state_remove, CmES5u32sYpV7JYN
+connection_state_remove, CP5puj4I8PtEU4qzYg
+connection_state_remove, C37jN32gN3y3AZzyf6
+connection_state_remove, C3eiCBGOLw3VtHfOj
+connection_state_remove, CwjjYJ2WqgTbAqiHl6
+connection_state_remove, C0LAHyvtKSQHyJxIl
+connection_state_remove, CFLRIC3zaTU1loLGxh
+connection_state_remove, C9rXSW3KSpTYvPrlI1
+connection_state_remove, Ck51lg1bScffFj34Ri
+connection_state_remove, C9mvWx3ezztgzcexV7
+connection_state_remove, CNnMIj2QSd84NKf7U3
+connection_state_remove, C7fIlMZDuRiqjpYbb
+connection_state_remove, CykQaM33ztNt0csB9a
+connection_state_remove, CtxTCR2Yer0FR1tIBg
+connection_state_remove, CpmdRlaUoJLN3uIRa
+connection_state_remove, C1Xkzz2MaGtLrc1Tla
+connection_state_remove, CqlVyW1YwZ15RhTBc4
+connection_state_remove, CLNN1k2QMum1aexUK7
+connection_state_remove, CBA8792iHmnhPLksKa
+connection_state_remove, CGLPPc35OzDQij1XX8
+connection_state_remove, CiyBAq1bBLNaTiTAc
+connection_state_remove, CFSwNi4CNGxcuffo49
+connection_state_remove, Cipfzj1BEnhejw8cGf
+connection_state_remove, CV5WJ42jPYbNW9JNWf
+connection_state_remove, CPhDKt12KQPUVbQz06
+connection_state_remove, CAnFrb2Cvxr5T7quOc
+connection_state_remove, C8rquZ3DjgNW06JGLl
+connection_state_remove, CzrZOtXqhwwndQva3
+connection_state_remove, CaGCc13FffXe6RkQl9
+epoch result
+remote:ff02::fb connections:1.0 orig_unique:1
+epoch result
+remote:208.80.152.2 connections:2.0 orig_unique:1
+epoch result
+remote:141.142.220.235 connections:1.0 orig_unique:1
+epoch result
+remote:141.142.220.255 connections:2.0 orig_unique:2
+epoch result
+remote:ff02::1:3 connections:2.0 orig_unique:1
+epoch result
+remote:141.142.2.2 connections:14.0 orig_unique:1
+epoch result
+remote:208.80.152.118 connections:1.0 orig_unique:1
+epoch result
+remote:208.80.152.3 connections:6.0 orig_unique:1
+epoch result
+remote:224.0.0.251 connections:3.0 orig_unique:3
+epoch result
+remote:224.0.0.252 connections:2.0 orig_unique:1
+epoch finished, 1300475173.475401
+zeek_done

testing/btest/scripts/base/frameworks/sumstats/final-epoch.zeek

@@ -0,0 +1,54 @@
+# @TEST-EXEC: zeek -b %INPUT -r $TRACES/wikipedia.trace
+# @TEST-EXEC: TEST_DIFF_CANONIFIER= btest-diff .stdout
+
+@load base/frameworks/sumstats
+
+event connection_state_remove(c: connection)
+	{
+	print "connection_state_remove", c$uid;
+	SumStats::observe("conn",
+		[$host=c$id$resp_h],
+		[$num=1]
+				);
+	SumStats::observe("orig_h",
+		[$host=c$id$resp_h],
+		[$str=cat(c$id$orig_h)]
+	);
+	}
+
+event zeek_init()
+	{
+	print "zeek_init";
+	SumStats::create([$name = "connections",
+		$epoch = 1hr,
+		$reducers = set(
+			SumStats::Reducer($stream="conn",
+				$apply=set(SumStats::SUM),
+			),
+			SumStats::Reducer($stream="orig_h",
+				$apply=set(SumStats::UNIQUE)
+			),
+		),
+		$epoch_finished(ts: time) = {
+			print "epoch finished", ts;
+		},
+		$epoch_result(ts: time, key: SumStats::Key, result: SumStats::Result) = {
+			print "epoch result";
+			print fmt("remote:%s connections:%s orig_unique:%s",
+			           key$host,
+			           result["conn"]$sum,
+			           result["orig_h"]$unique);
+		}
+	]);
+	}
+
+event net_done(ts: time)
+	{
+	print "net_done", ts;
+	}
+
+
+event zeek_done()
+	{
+	print "zeek_done";
+	}