make openflow framework work in clusters.

scripts/base/frameworks/openflow/__load__.bro

@@ -2,3 +2,12 @@
 @load ./types
 @load ./main
 @load ./plugins
+
+# The cluster framework must be loaded first.
+@load base/frameworks/cluster
+
+@if ( Cluster::is_enabled() )
+@load ./cluster
+@else
+@load ./non-cluster
+@endif

scripts/base/frameworks/openflow/cluster.bro

@@ -0,0 +1,94 @@
+@load ./main
+@load base/frameworks/cluster
+
+module OpenFlow;
+
+export {
+	## This is the event used to transport flow_mod messages to the manager.
+	global cluster_flow_mod: event(name: string, match: ofp_match, flow_mod: ofp_flow_mod);
+
+	## This is the event used to transport flow_clear messages to the manager.
+	global cluster_flow_clear: event(name: string);
+}
+
+## Workers need ability to forward commands to manager.
+redef Cluster::worker2manager_events += /OpenFlow::cluster_flow_(mod|clear)/;
+
+global name_to_controller: table[string] of Controller;
+
+# the flow_mod function wrapper
+function flow_mod(controller: Controller, match: ofp_match, flow_mod: ofp_flow_mod): bool
+	{
+	if ( ! controller?$flow_mod )
+		return F;
+
+	if ( Cluster::local_node_type() == Cluster::MANAGER )
+		return controller$flow_mod(controller$state, match, flow_mod);
+	else
+		event OpenFlow::cluster_flow_mod(controller$state$_name, match, flow_mod);
+
+	return T;
+	}
+
+function flow_clear(controller: Controller): bool
+	{
+	if ( ! controller?$flow_clear )
+		return F;
+
+	if ( Cluster::local_node_type() == Cluster::MANAGER )
+		return controller$flow_clear(controller$state);
+	else
+		event OpenFlow::cluster_flow_clear(controller$state$_name);
+
+	return T;
+	}
+
+@if ( Cluster::local_node_type() == Cluster::MANAGER )
+event OpenFlow::cluster_flow_mod(name: string, match: ofp_match, flow_mod: ofp_flow_mod)
+	{
+	if ( name !in name_to_controller )
+		{
+		Reporter::error(fmt("OpenFlow controller %s not found in mapping on master", name));
+		return;
+		}
+
+	local c = name_to_controller[name];
+	if ( c?$flow_mod )
+		c$flow_mod(c$state, match, flow_mod);
+	}
+
+event OpenFlow::cluster_flow_clear(name: string)
+	{
+	if ( name !in name_to_controller )
+		{
+		Reporter::error(fmt("OpenFlow controller %s not found in mapping on master", name));
+		return;
+		}
+
+	local c = name_to_controller[name];
+
+	if ( c?$flow_clear )
+		c$flow_clear(c$state);
+	}
+@endif
+
+function register_controller(tpe: OpenFlow::Plugin, name: string, controller: Controller)
+	{
+	controller$state$_name = cat(tpe, name);
+	controller$state$_plugin = tpe;
+
+	# we only run the init functions on the manager.
+	if ( Cluster::local_node_type() != Cluster::MANAGER )
+		return;
+
+	if ( controller$state$_name in name_to_controller )
+		{
+		Reporter::error("OpenFlow Controller %s was already registered. Ignored duplicate registration");
+		return;
+		}
+
+	name_to_controller[controller$state$_name] = controller;
+
+	if ( controller?$init )
+		controller$init(controller$state);
+	}

scripts/base/frameworks/openflow/main.bro

@@ -80,7 +80,7 @@ export {
 	global match_conn: function(id: conn_id, reverse: bool &default=F): ofp_match;
 
 	# ###
-	# ### Low-level functions for cookie handling.
+	# ### Low-level functions for cookie handling and plugin registration.
 	# ###
 
 	## Function to get the unique id out of a given cookie.
@@ -103,26 +103,18 @@ export {
 	##
 	## Returns: The cookie group id.
 	global generate_cookie: function(cookie: count &default=0): count;
+
+	## Function to register a controller instance. This function
+	## is called automatically by the plugin _new functions.
+	##
+	## tpe: type of this plugin
+	##
+	## name: unique name of this controller instance.
+	##
+	## controller: The controller to register
+	global register_controller: function(tpe: OpenFlow::Plugin, name: string, controller: Controller);
 }
 
-
-# the flow_mod function wrapper
-function flow_mod(controller: Controller, match: ofp_match, flow_mod: ofp_flow_mod): bool
-	{
-		if ( controller?$flow_mod )
-			return controller$flow_mod(controller$state, match, flow_mod);
-		else
-			return F;
-	}
-
-function flow_clear(controller: Controller): bool
-	{
-	if ( controller?$flow_clear )
-		return controller$flow_clear(controller$state);
-	else
-		return F;
-	}
-
 function match_conn(id: conn_id, reverse: bool &default=F): ofp_match
 	{
 	local dl_type = ETH_IPv4;

scripts/base/frameworks/openflow/non-cluster.bro

@@ -0,0 +1,29 @@
+@load ./main
+
+module OpenFlow;
+
+# the flow_mod function wrapper
+function flow_mod(controller: Controller, match: ofp_match, flow_mod: ofp_flow_mod): bool
+	{
+	if ( controller?$flow_mod )
+		return controller$flow_mod(controller$state, match, flow_mod);
+	else
+		return F;
+	}
+
+function flow_clear(controller: Controller): bool
+	{
+	if ( controller?$flow_clear )
+		return controller$flow_clear(controller$state);
+	else
+		return F;
+	}
+
+function register_controller(tpe: OpenFlow::Plugin, name: string, controller: Controller)
+	{
+	controller$state$_name = cat(tpe, name);
+	controller$state$_plugin = tpe;
+
+	if ( controller?$init )
+		controller$init(controller$state);
+	}

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

@@ -19,7 +19,7 @@ export {
 	## dpid: OpenFlow switch datapath id.
 	##
 	## Returns: OpenFlow::Controller record
-	global broker_new: function(host: addr, host_port: port, topic: string, dpid: count): OpenFlow::Controller;
+	global broker_new: function(name: string, host: addr, host_port: port, topic: string, dpid: count): OpenFlow::Controller;
 
 	redef record ControllerState += {
 		## Controller ip.
@@ -55,14 +55,21 @@ function broker_flow_clear_fun(state: OpenFlow::ControllerState): bool
 	return T;
 	}
 
-# broker controller constructor
-function broker_new(host: addr, host_port: port, topic: string, dpid: count): OpenFlow::Controller
+function broker_init(state: OpenFlow::ControllerState)
 	{
 	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, $supports_flow_removed=T];
+	BrokerComm::connect(cat(state$broker_host), state$broker_port, 1sec);
+	BrokerComm::subscribe_to_events(state$broker_topic); # openflow success and failure events are directly sent back via the other plugin via broker.
+	}
+
+# broker controller constructor
+function broker_new(name: string, host: addr, host_port: port, topic: string, dpid: count): OpenFlow::Controller
+	{
+	local c = OpenFlow::Controller($state=OpenFlow::ControllerState($broker_host=host, $broker_port=host_port, $broker_dpid=dpid, $broker_topic=topic),
+		$flow_mod=broker_flow_mod_fun, $flow_clear=broker_flow_clear_fun, $describe=broker_describe, $supports_flow_removed=T, $init=broker_init);
+
+	register_controller(OpenFlow::BROKER, name, c);
+
+	return c;
 	}
 

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

@@ -67,6 +67,10 @@ function log_describe(state: ControllerState): string
 
 function log_new(dpid: count, success_event: bool &default=T): OpenFlow::Controller
 	{
-	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, $supports_flow_removed=F];
+	local c = OpenFlow::Controller($state=OpenFlow::ControllerState($log_dpid=dpid, $log_success_event=success_event),
+		$flow_mod=log_flow_mod, $flow_clear=ryu_flow_clear, $describe=log_describe, $supports_flow_removed=F);
+
+	register_controller(OpenFlow::OFLOG, cat(dpid), c);
+
+	return c;
 	}

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

@@ -179,6 +179,10 @@ function ryu_describe(state: ControllerState): string
 # Ryu controller constructor
 function ryu_new(host: addr, host_port: count, dpid: count): OpenFlow::Controller
 	{
-	return [$state=[$ryu_host=host, $ryu_port=host_port, $ryu_dpid=dpid, $_plugin=OpenFlow::RYU],
-		$flow_mod=ryu_flow_mod, $flow_clear=ryu_flow_clear, $describe=ryu_describe, $supports_flow_removed=F];
+	local c = OpenFlow::Controller($state=OpenFlow::ControllerState($ryu_host=host, $ryu_port=host_port, $ryu_dpid=dpid),
+		$flow_mod=ryu_flow_mod, $flow_clear=ryu_flow_clear, $describe=ryu_describe, $supports_flow_removed=F);
+
+	register_controller(OpenFlow::RYU, cat(host,host_port,dpid), c);
+
+	return c;
 	}

scripts/base/frameworks/openflow/types.bro

@@ -13,8 +13,10 @@ export {
 	## Can be redefined by plugins to
 	## add state.
 	type ControllerState: record {
-		## Internally set to the plugin used.
-		_plugin: Plugin;
+		## Internally set to the type of plugin used.
+		_plugin: Plugin &optional;
+		## Internally set to the unique name of the controller.
+		_name: string &optional;
 	} &redef;
 
 	## Openflow match definition.
@@ -143,6 +145,8 @@ export {
 		supports_flow_removed: bool;
 		## function that describes the controller. Has to be implemented.
 		describe: function(state: ControllerState): string;
+		## one-time initialization function.
+		init: function (state: ControllerState) &optional;
 		## flow_mod function
 		flow_mod: function(state: ControllerState, match: ofp_match, flow_mod: ofp_flow_mod): bool &optional;
 		## flow_clear function