Source file OpenFlow0x04_Plugin.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
open Core
open Async

(* Marshal and send a message to the switch *)
let send_message (to_client : Writer.t) (xid : Frenetic_kernel.OpenFlow_Header.xid)
  (message : Frenetic_kernel.OpenFlow0x04.Message.t) : unit =
  let raw_message = Frenetic_kernel.OpenFlow0x04.Message.marshal xid message in
  Writer.write to_client raw_message

(* Send group messages to switch to make group table *)
let implement_group_table (writer : Writer.t) (tbl : Frenetic_kernel.GroupTable0x04.t) : unit =
  let msgs = Frenetic_kernel.GroupTable0x04.commit tbl in
  let msg_num = List.length msgs in
  List.iteri msgs ~f:(fun i msg -> send_message writer (Int32.of_int_exn (9000 + i)) msg);
  if msg_num <> 0 then
    Logging.info "Sent %d Group Table message(s)" (List.length msgs)

(* Add mask so that the meta value can be changed *)
let mask_meta (meta_id : int) =
  Frenetic_kernel.OpenFlow0x04.{ m_value = Int64.of_int meta_id; m_mask = Some 64L }

(* Send FlowMod messages to switch to implement policy *)
let implement_flow (writer : Writer.t) (fdd : Frenetic_netkat.Local_compiler.t)
  (layout : Frenetic_netkat.Local_compiler.flow_layout)
  (sw_id : Frenetic_kernel.OpenFlow.switchId) : unit =
  let open Frenetic_kernel.OpenFlow0x04 in
  let open Frenetic_netkat.Local_compiler in
  let (flow_rows, group_tbl) = to_multitable sw_id layout fdd in
  implement_group_table writer group_tbl;
  List.iteri flow_rows ~f:(fun i row ->
    let (tbl, m_id) = row.flowId in
    let xid = Int32.of_int_exn i in
    let prio = 1000 - i in
    (* do not include meta value for table start, else everything drops *)
    let pat = if m_id = 0 then (Oxm.from_of_pattern row.pattern)
      else (OxmMetadata (mask_meta m_id))::(Oxm.from_of_pattern row.pattern) in
    (* Must order prereq  *)
    let pat_reversed = List.rev pat in
    let insts = match row.instruction with
      | `Action action_group -> Instructions.from_of_group action_group
      | `GotoTable (goto_t, goto_m) ->
        [WriteMetadata (mask_meta goto_m); GotoTable goto_t]
    in
    let message = Message.FlowModMsg (add_flow ~tbl ~prio ~pat:pat_reversed ~insts) in
    Logging.info "Sending flow to switch %Ld\n\ttable:%d\n\tpriority:%d\n\tpattern:%s\n\tinstructions:%s"
      sw_id tbl prio (Oxm.match_to_string pat_reversed) (Instructions.to_string insts);
    send_message writer xid message)

(* Send FlowMod messages to switch to implement the policy, use topology to
 * generate fault tolerant group tables. *)
let implement_tolerant_flow (writer : Writer.t) (fdd : Frenetic_netkat.Local_compiler.t)
  (topo : Frenetic_kernel.Net.Net.Topology.t) (sw_id : Frenetic_kernel.OpenFlow.switchId)
  : unit =
  let open Frenetic_kernel.OpenFlow0x04 in
  let flowtable = Frenetic_netkat.Local_compiler.to_table sw_id fdd in
  List.iteri flowtable ~f:(fun i row ->
    let tbl = 1 in
    let xid= Int32.of_int_exn i in
    let prio = 1000 - i in
    let pat = Oxm.from_of_pattern row.pattern in
    (* Must order prereq  *)
    let pat_reversed = List.rev pat in
    let insts = Instructions.from_of_group row.action in
    let message = Message.FlowModMsg (add_flow ~tbl ~prio ~pat:pat_reversed ~insts) in
    Logging.info "Sending flow to switch %Ld\n\ttable:%d\n\tpriority:%d\n\tpattern:%s\n\tinstructions:%s"
      sw_id tbl prio (Oxm.match_to_string pat_reversed) (Instructions.to_string insts);
    send_message writer xid message)

(* Respond to message from switch *)
let process_message (xid : Frenetic_kernel.OpenFlow_Header.xid) (message : Frenetic_kernel.OpenFlow0x04.Message.t)
  (message_sender : (Frenetic_kernel.OpenFlow_Header.xid -> Frenetic_kernel.OpenFlow0x04.Message.t -> unit))
  (flow_sender : Frenetic_kernel.OpenFlow.switchId -> unit) : unit =
  let open Frenetic_kernel.OpenFlow0x04 in
  match message with
  | Message.EchoRequest bytes -> message_sender xid (Message.EchoReply bytes)
  | Message.Hello _           -> message_sender 10l Message.FeaturesRequest
  | Message.FeaturesReply fts -> flow_sender fts.datapath_id
  | Message.Error error       -> Logging.error "%s" (Error.to_string error)
  | _                         -> Logging.info "Unsupported message type"

(* Parse incoming client messages and respond. `Finished is sent if an
 * error occurs, otherwise `Repeat indefinitely. *)
let read_respond_loop (reader : Reader.t)
  (message_sender : (Frenetic_kernel.OpenFlow_Header.xid -> Frenetic_kernel.OpenFlow0x04.Message.t -> unit))
  (flow_sender : Frenetic_kernel.OpenFlow.switchId -> unit) ()
  : [ `Finished of unit | `Repeat of unit ] Deferred.t =
  let header_buf = Bytes.create Frenetic_kernel.OpenFlow_Header.size in
  Reader.really_read reader header_buf
  >>= function
  | `Eof _ ->
    Logging.info "Connection closed reading header";
    return (`Finished ())
  | `Ok ->
    let header = Frenetic_kernel.OpenFlow_Header.parse (Cstruct.of_bytes header_buf) in
    let message_len = header.length - Frenetic_kernel.OpenFlow_Header.size in
    let message_buf = Bytes.create message_len in
    Reader.really_read reader message_buf
    >>= function
    | `Eof _ ->
      Logging.info "Error reading client message";
      return (`Finished ())
    | `Ok ->
      let (xid, body) = Frenetic_kernel.OpenFlow0x04.Message.parse header (Bytes.to_string message_buf) in
      process_message xid body message_sender flow_sender;
      return (`Repeat ())

(* Send the initil handshake, loop on client response *)
let client_handler (reader : Reader.t)
  (message_sender : (Frenetic_kernel.OpenFlow_Header.xid -> Frenetic_kernel.OpenFlow0x04.Message.t -> unit))
  (flow_sender : Frenetic_kernel.OpenFlow.switchId -> unit) : unit Deferred.t =
  Logging.info "Client connected";
  message_sender 0l (Frenetic_kernel.OpenFlow0x04.Message.Hello [VersionBitMap [0x04]]);
  Logging.info "Sent Hello";
  Deferred.repeat_until_finished () (read_respond_loop reader message_sender flow_sender)

(* Implement multi-table policies. Extract the policy from a kat file,
 * run client handler for each connecting client *)
let main (of_port : int) (pol_file : string)
  (layout : Frenetic_netkat.Local_compiler.flow_layout) () : unit =
  let open Frenetic_netkat.Local_compiler in
  Logging.info "Starting OpenFlow 1.3 controller";
  Logging.info "Using flow tables: %s" (layout_to_string layout);
  let pol = Frenetic_netkat.Parser.pol_of_file pol_file in
  let compiler_opts = {default_compiler_options with field_order = `Static (List.concat layout)} in
  let fdd = compile pol ~options:compiler_opts in
  let _ = Tcp.Server.create ~on_handler_error:`Raise (Tcp.Where_to_listen.of_port of_port)
    (fun _ reader writer ->
      let message_sender = send_message writer in
      let flow_sender = implement_flow writer fdd layout in
      client_handler reader message_sender flow_sender)
  in ()

(* Implement fault tolerant policies. Extract the policy and topology from
 * kat and dot files, run client_handler for each connecting client
 * TODO(mulias): This is a SHAM. Parsing the topology from a .dot file is not
 * yet implemented. *)
let fault_tolerant_main (of_port : int) (pol_file : string)
  (topo_file : string) () : unit =
  Logging.info "Starting OpenFlow 1.3 fault tolerant controller";
  let pol = Frenetic_netkat.Parser.pol_of_file pol_file in
  let fdd = Frenetic_netkat.Local_compiler.compile pol in
  let topo = Frenetic_kernel.Net.Net.Topology.empty () in
  (* let topo = Frenetic_kernel.Net.Net.Parse.from_dotfile topo_file in *)
  let _ = Tcp.Server.create ~on_handler_error:`Raise (Tcp.Where_to_listen.of_port of_port)
    (fun _ reader writer ->
      let message_sender = send_message writer in
      let flow_sender = implement_tolerant_flow writer fdd topo in
      client_handler reader message_sender flow_sender)
  in ()