handle the notification events correctly.

Now if a rule is inserted correctly (or fails to be inserted) into
openflow, we actually get the corresponding Pacf events that everything
worked.

scripts/base/frameworks/openflow/main.bro

@@ -38,7 +38,7 @@ export {
 	##
 	## flow_mod: The openflow flow_mod record which describes the action to take.
 	##
-	## msg: An optional informational message by the plugin..
+	## msg: An optional informational message by the plugin.
 	global flow_mod_success: event(match: ofp_match, flow_mod: ofp_flow_mod, msg: string &default="");
 
 	## Reports an error while installing a flow Rule.

scripts/base/frameworks/openflow/plugins/broker.bro

@@ -60,6 +60,7 @@ function broker_new(host: addr, host_port: port, topic: string, dpid: count): Op
 	{
 	BrokerComm::enable();
 	BrokerComm::connect(cat(host), host_port, 1sec);
+	BrokerComm::subscribe_to_events(topic); # openflow success and failure events are directly sent back via the other plugin via broker.
 
 	return [$state=[$broker_host=host, $broker_port=host_port, $broker_dpid=dpid, $broker_topic=topic, $_plugin=OpenFlow::BROKER],
 		$flow_mod=broker_flow_mod_fun, $flow_clear=broker_flow_clear_fun, $describe=broker_describe];

scripts/base/frameworks/openflow/plugins/log.bro

@@ -17,12 +17,16 @@ export {
 	##
 	## dpid: OpenFlow switch datapath id.
 	##
+	## success_event: If true, flow_mod_success is raised for each logged line.
+	##
 	## Returns: OpenFlow::Controller record
-	global log_new: function(dpid: count): OpenFlow::Controller;
+	global log_new: function(dpid: count, success_event: bool &default=T): OpenFlow::Controller;
 
 	redef record ControllerState += {
 		## OpenFlow switch datapath id.
 		log_dpid: count &optional;
+		## Raise or do not raise success event
+		log_success_event: bool &optional;
 	};
 
 	## The record type which contains column fields of the OpenFlow log.
@@ -50,6 +54,8 @@ event bro_init() &priority=5
 function log_flow_mod(state: ControllerState, match: ofp_match, flow_mod: OpenFlow::ofp_flow_mod): bool
 	{
 	Log::write(OpenFlow::LOG, [$ts=network_time(), $dpid=state$log_dpid, $match=match, $flow_mod=flow_mod]);
+	if ( state$log_success_event )
+		event OpenFlow::flow_mod_success(match, flow_mod);
 
 	return T;
 	}
@@ -59,8 +65,8 @@ function log_describe(state: ControllerState): string
 	return fmt("OpenFlog Log Plugin - DPID %d", state$log_dpid);
 	}
 
-function log_new(dpid: count): OpenFlow::Controller
+function log_new(dpid: count, success_event: bool &default=T): OpenFlow::Controller
 	{
-	return [$state=[$log_dpid=dpid, $_plugin=OpenFlow::LOG],
+	return [$state=[$log_dpid=dpid, $log_success_event=success_event, $_plugin=OpenFlow::LOG],
 		$flow_mod=log_flow_mod, $flow_clear=ryu_flow_clear, $describe=log_describe];
 	}

scripts/base/frameworks/pacf/plugins/openflow.bro

@@ -23,10 +23,32 @@ export {
 		of_config: OfConfig &optional;
 	};
 
+	type OfTable: record {
+		p: PluginState;
+		r: Rule;
+	};
+
+	## the time interval after which an openflow message is considered to be timed out
+	## and we delete it from our internal tracking.
+	const openflow_timeout = 20secs &redef;
+
 	## Instantiates an openflow plugin for the PACF framework.
 	global create_openflow: function(controller: OpenFlow::Controller, config: OfConfig &default=[]) : PluginState;
 }
 
+global of_messages: table[count, OpenFlow::ofp_flow_mod_command] of OfTable &create_expire=openflow_timeout
+	&expire_func=function(t: table[count, OpenFlow::ofp_flow_mod_command] of OfTable, idx: any): interval
+		{
+		local rid: count;
+		local command: OpenFlow::ofp_flow_mod_command;
+		[rid, command] = idx;
+
+		local p = t[rid, command]$p;
+		local r = t[rid, command]$r;
+		event Pacf::rule_error(r, p, "Timeout during rule insertion/removal");
+		return 0secs;
+		};
+
 function openflow_name(p: PluginState) : string
 	{
 	return fmt("Openflow - %s", p$of_controller$describe(p$of_controller$state));
@@ -177,7 +199,7 @@ function openflow_rule_to_flow_mod(p: PluginState, r: Rule) : OpenFlow::ofp_flow
 	local c = p$of_config;
 
 	local flow_mod = OpenFlow::ofp_flow_mod(
-		$cookie=r$id,
+		$cookie=OpenFlow::generate_cookie(r$id),
 		$command=OpenFlow::OFPFC_ADD,
 		$idle_timeout=c$idle_timeout,
 		$priority=int_to_count(r$priority + c$priority_offset)
@@ -220,7 +242,9 @@ function openflow_add_rule(p: PluginState, r: Rule) : bool
 
 	for ( i in matches )
 		{
-		if ( ! OpenFlow::flow_mod(p$of_controller, matches[i], flow_mod) )
+		if ( OpenFlow::flow_mod(p$of_controller, matches[i], flow_mod) )
+			of_messages[r$id, flow_mod$command] = OfTable($p=p, $r=r);
+		else
 			event rule_error(r, p, "Error while executing OpenFlow::flow_mod");
 		}
 
@@ -233,15 +257,47 @@ function openflow_remove_rule(p: PluginState, r: Rule) : bool
 		return F;
 
 	local flow_mod: OpenFlow::ofp_flow_mod = [
-		$cookie=r$id,
+		$cookie=OpenFlow::generate_cookie(r$id),
 		$command=OpenFlow::OFPFC_DELETE
 	];
 
-	OpenFlow::flow_mod(p$of_controller, [], flow_mod);
+	if ( OpenFlow::flow_mod(p$of_controller, [], flow_mod) )
+			of_messages[r$id, flow_mod$command] = OfTable($p=p, $r=r);
+	else
+			event rule_error(r, p, "Error while executing OpenFlow::flow_mod");
 
 	return T;
 	}
 
+event OpenFlow::flow_mod_success(match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod, msg: string) &priority=3
+	{
+	local id = OpenFlow::get_cookie_uid(flow_mod$cookie);
+	if ( [id, flow_mod$command] !in of_messages )
+		return;
+
+	local r = of_messages[id,flow_mod$command]$r;
+	local p = of_messages[id,flow_mod$command]$p;
+	delete of_messages[id,flow_mod$command];
+
+	if ( flow_mod$command == OpenFlow::OFPFC_ADD )
+		event Pacf::rule_added(r, p, msg);
+	else if ( flow_mod$command == OpenFlow::OFPFC_DELETE || flow_mod$command == OpenFlow::OFPFC_DELETE_STRICT )
+		event Pacf::rule_removed(r, p, msg);
+	}
+
+event OpenFlow::flow_mod_failure(match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod, msg: string) &priority=3
+	{
+	local id = OpenFlow::get_cookie_uid(flow_mod$cookie);
+	if ( [id, flow_mod$command] !in of_messages )
+		return;
+
+	local r = of_messages[id,flow_mod$command]$r;
+	local p = of_messages[id,flow_mod$command]$p;
+	delete of_messages[id,flow_mod$command];
+
+	event Pacf::rule_error(r, p, msg);
+	}
+
 global openflow_plugin = Plugin(
 	$name=openflow_name,
 	$can_expire = T,

testing/btest/Baseline/scripts.base.frameworks.openflow.broker-basic/send.send.out

@@ -1,2 +1,8 @@
 BrokerComm::outgoing_connection_established, 127.0.0.1, 9999/tcp
+Flow_mod_success
+Flow_mod_failure
 connection established
+Flow_mod_success
+Flow_mod_failure
+Flow_mod_success
+Flow_mod_failure

testing/btest/Baseline/scripts.base.frameworks.openflow.ryu-basic/.stdout

@@ -1,7 +1,10 @@
 http://127.0.0.1:8080/stats/flowentry/clear/42
 http://127.0.0.1:8080/stats/flowentry/add
 {"match": {}, "dpid": 42, "flags": 0, "hard_timeout": 0, "priority": 0, "actions": [{"port": 3, "type": "OUTPUT"}, {"port": 7, "type": "OUTPUT"}], "cookie": 4398046511105, "idle_timeout": 0}
+Flow_mod_success
 http://127.0.0.1:8080/stats/flowentry/add
 {"match": {"tp_dst": 25, "nw_dst": "74.53.140.153/32", "nw_src": "10.10.1.4/32", "dl_type": 2048, "tp_src": 1470, "nw_proto": 6}, "dpid": 42, "flags": 0, "hard_timeout": 0, "priority": 5, "actions": [], "cookie": 4398046511146, "idle_timeout": 30}
 http://127.0.0.1:8080/stats/flowentry/add
 {"match": {"tp_dst": 25, "nw_dst": "10.10.1.4/32", "nw_src": "74.53.140.153/32", "dl_type": 2048, "tp_src": 25, "nw_proto": 6}, "dpid": 42, "flags": 0, "hard_timeout": 0, "priority": 5, "actions": [], "cookie": 4398046511146, "idle_timeout": 30}
+Flow_mod_success
+Flow_mod_success

testing/btest/Baseline/scripts.base.frameworks.pacf.openflow/openflow.log

@@ -3,10 +3,10 @@
 #empty_field	(empty)
 #unset_field	-
 #path	openflow
-#open	2015-04-15-19-15-14
+#open	2015-05-15-17-45-52
 #fields	ts	dpid	match.in_port	match.dl_src	match.dl_dst	match.dl_vlan	match.dl_vlan_pcp	match.dl_type	match.nw_tos	match.nw_proto	match.nw_src	match.nw_dst	match.tp_src	match.tp_dst	flow_mod.cookie	flow_mod.table_id	flow_mod.command	flow_mod.idle_timeout	flow_mod.hard_timeout	flow_mod.priority	flow_mod.out_port	flow_mod.out_group	flow_mod.flags	flow_mod.out_ports
 #types	time	count	count	string	string	count	count	count	count	count	subnet	subnet	port	port	count	count	enum	count	count	count	count	count	count	vector[count]
-1254722767.875996	42	-	-	-	-	-	2048	-	6	10.10.1.4/32	74.53.140.153/32	1470	25	2	-	OpenFlow::OFPFC_ADD	60	30	0	-	-	0	(empty)
+1254722767.875996	42	-	-	-	-	-	2048	-	6	10.10.1.4/32	74.53.140.153/32	1470	25	4398046511106	-	OpenFlow::OFPFC_ADD	60	30	0	-	-	0	(empty)
-1254722767.875996	42	-	-	-	-	-	2048	-	-	10.10.1.4/32	-	-	-	3	-	OpenFlow::OFPFC_ADD	60	15	0	-	-	0	(empty)
+1254722767.875996	42	-	-	-	-	-	2048	-	-	10.10.1.4/32	-	-	-	4398046511107	-	OpenFlow::OFPFC_ADD	60	15	0	-	-	0	(empty)
-1254722767.875996	42	-	-	-	-	-	2048	-	-	-	10.10.1.4/32	-	-	3	-	OpenFlow::OFPFC_ADD	60	15	0	-	-	0	(empty)
+1254722767.875996	42	-	-	-	-	-	2048	-	-	-	10.10.1.4/32	-	-	4398046511107	-	OpenFlow::OFPFC_ADD	60	15	0	-	-	0	(empty)
-#close	2015-04-15-19-15-14
+#close	2015-05-15-17-45-53

testing/btest/Baseline/scripts.base.frameworks.pacf.openflow/pacf.log

@@ -3,10 +3,12 @@
 #empty_field	(empty)
 #unset_field	-
 #path	pacf
-#open	2015-05-12-20-36-53
+#open	2015-05-15-18-21-40
 #fields	ts	category	cmd	state	action	target	entity_type	entity	msg	location	plugin
 #types	time	enum	string	enum	string	enum	string	string	string	string	string
 0.000000	Pacf::MESSAGE	-	-	-	-	-	-	activated plugin with priority 0	-	Openflow - OpenFlog Log Plugin - DPID 42
 1254722767.875996	Pacf::RULE	ADD	Pacf::REQUESTED	Pacf::DROP	Pacf::MONITOR	Pacf::FLOW	10.10.1.4/32/1470->74.53.140.153/32/25	-	(empty)	Openflow - OpenFlog Log Plugin - DPID 42
 1254722767.875996	Pacf::RULE	ADD	Pacf::REQUESTED	Pacf::DROP	Pacf::FORWARD	Pacf::ADDRESS	10.10.1.4/32	-	(empty)	Openflow - OpenFlog Log Plugin - DPID 42
-#close	2015-05-12-20-36-53
+1254722767.875996	Pacf::RULE	ADD	Pacf::SUCCEEDED	Pacf::DROP	Pacf::MONITOR	Pacf::FLOW	10.10.1.4/32/1470->74.53.140.153/32/25	-	(empty)	Openflow - OpenFlog Log Plugin - DPID 42
+1254722767.875996	Pacf::RULE	ADD	Pacf::SUCCEEDED	Pacf::DROP	Pacf::FORWARD	Pacf::ADDRESS	10.10.1.4/32	-	(empty)	Openflow - OpenFlog Log Plugin - DPID 42
+#close	2015-05-15-18-21-40

testing/btest/scripts/base/frameworks/openflow/broker-basic.bro

@@ -56,6 +56,15 @@ event connection_established(c: connection)
 	OpenFlow::flow_mod(of_controller, match_rev, flow_mod);
 	}
 
+event OpenFlow::flow_mod_success(match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod, msg: string)
+	{
+	print "Flow_mod_success";
+	}
+
+event OpenFlow::flow_mod_failure(match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod, msg: string)
+	{
+	print "Flow_mod_failure";
+	}
 
 @TEST-END-FILE
 
@@ -91,6 +100,8 @@ function got_message()
 event OpenFlow::broker_flow_mod(dpid: count, match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod)
 	{
 	print "got flow_mod", dpid, match, flow_mod;
+	BrokerComm::event("bro/event/openflow", BrokerComm::event_args(OpenFlow::flow_mod_success, match, flow_mod, ""));
+	BrokerComm::event("bro/event/openflow", BrokerComm::event_args(OpenFlow::flow_mod_failure, match, flow_mod, ""));
 	got_message();
 	}
 

testing/btest/scripts/base/frameworks/openflow/ryu-basic.bro

@@ -30,3 +30,8 @@ event connection_established(c: connection)
 	OpenFlow::flow_mod(of_controller, match, flow_mod);
 	OpenFlow::flow_mod(of_controller, match_rev, flow_mod);
 	}
+
+event OpenFlow::flow_mod_success(match: OpenFlow::ofp_match, flow_mod: OpenFlow::ofp_flow_mod, msg: string)
+	{
+	print "Flow_mod_success";
+	}