Async/streaming gRPC client/server proto
This experiment was to fine-tune how we can implement
async gRPC client and server code inside a Twisted
python app.
Change-Id: I945014e27f4b9d6ed624666e0284cc298548adb3
Major cleanup of openflow_13.proto
Change-Id: I4e54eaf87b682124ec518a0ade1a6050a6ec6da8
Relocated openflow_13.proto to voltha
Change-Id: I66ae45a9142d180c2c6651e75c7a1ee08aef7ef8
Removed forced utest from make build
Change-Id: If0da58e9d135ebde6ca68c3316688a03a7b10f2f
twisted openflow agent first pass
Change-Id: Ibe5b4727ccfe92e6fd464ccd3baf6275569ef5d3
store openflow derived files
Change-Id: Ib3e1384bb2ca2a9c0872767f7b793f96b0a154e2
Minor cleanup
Change-Id: I1280ed3acb606121b616a0efd573f5f59d010dca
Factored out common utils
Change-Id: Icd86fcd50f60d0900924674cbcd65e13e47782a1
Refactored twisted agent
Change-Id: I71f26ce5357a4f98477df60b8c5ddc068cf75d43
Relocated openflow agent to ofagent
... and preserved obsolete working (non-twisted) agent under
~/obsolete, so we can still run the olt-oftest and pass tests,
unit the new twisted based agent reaches that maturity point.
Change-Id: I727f8d7144b1291a40276dad2966b7643bd7bc4b
olt-oftest in fake mode works with new agent
Change-Id: I43b4f5812e8dfaa9f45e4a77fdcf6c30ac520f8d
Initial ofagent/voltha operation
Change-Id: Ia8104f1285a6b1c51635d36d7d78fc113f800e79
Additional callouts to Voltha
Change-Id: If8f483d5140d3c9d45f22b480b8d33249a29cd4e
More gRPC calls
Change-Id: I7d24fadf9425217fb26ffe18f25359d072ef38fa
Flow add/list now works
Change-Id: Ie3e3e73108645b47891cef798fc61372a022fd93
Missed some files
Change-Id: I29e81238ff1a26c095c0c73e521579edf7092e21
diff --git a/obsolete/store.py b/obsolete/store.py
new file mode 100644
index 0000000..0bd04d6
--- /dev/null
+++ b/obsolete/store.py
@@ -0,0 +1,335 @@
+"""
+Internal state of the device
+"""
+
+import logging
+from ofagent.utils import pp
+import loxi.of13 as ofp
+
+
+class GroupEntry(object):
+ def __init__(self, group_desc, group_stats):
+ assert isinstance(group_desc, ofp.group_desc_stats_entry)
+ assert isinstance(group_stats, ofp.group_stats_entry)
+ self.group_desc = group_desc
+ self.group_stats = group_stats
+
+
+def flow_stats_entry_from_flow_mod_message(fmm):
+ assert isinstance(fmm, ofp.message.flow_mod)
+
+ # extract a flow stats entry from a flow_mod message
+ kw = fmm.__dict__
+
+ # drop these from the object
+ for k in ('xid', 'cookie_mask', 'out_port', 'buffer_id', 'out_group'):
+ del kw[k]
+ flow = ofp.flow_stats_entry(
+ duration_sec=0,
+ duration_nsec=0,
+ packet_count=0,
+ byte_count=0,
+ **kw
+ )
+ return flow
+
+def group_entry_from_group_mod_message(gmm):
+ assert isinstance(gmm, ofp.message.group_mod)
+
+ kw = gmm.__dict__
+
+ # drop these attributes from the object:
+ for k in ('xid',):
+ del kw[k]
+
+ group_desc = ofp.group_desc_stats_entry(
+ group_type=gmm.group_type,
+ group_id=gmm.group_id,
+ buckets=gmm.buckets
+ )
+
+ group_stats = ofp.group_stats_entry(
+ group_id=gmm.group_id
+ )
+
+ return GroupEntry(group_desc, group_stats)
+
+
+class ObjectStore(object):
+
+ def __init__(self):
+ self.ports = [] # list of ofp.common.port_desc
+ self.flows = [] # list of ofp.common.flow_stats_entry
+ self.groups = {} # dict of (ofp.group_desc_stats_entry, ofp.group_stats_entry) tuples,
+ # keyed by the group_id field
+ self.agent = None
+
+ def set_agent(self, agent):
+ """Set agent reference"""
+ self.agent = agent
+
+ def signal_flow_mod_error(self, code, flow_mod):
+ """Forward error to agent"""
+ if self.agent is not None:
+ agent.signal_flow_mod_error(code, flow_mod)
+
+ def signal_flow_removal(self, flow):
+ """Forward flow removal notification to agent"""
+ if self.agent is not None:
+ agent.signal_flow_removal(flow)
+
+ def signal_group_mod_error(self, code, group_mod):
+ if self.agent is not None:
+ agent.signal_group_mod_error(code, group_mod)
+
+ ## <=========================== PORT HANDLERS ==================================>
+
+ def port_list(self):
+ return self.ports
+
+ def port_add(self, port):
+ self.ports.append(port)
+
+ def port_stats(self):
+ logging.warn("port_stats() not yet implemented")
+ return []
+
+ ## <=========================== FLOW HANDLERS ==================================>
+
+ def flow_add(self, flow_add):
+ assert isinstance(flow_add, ofp.message.flow_add)
+ assert flow_add.cookie_mask == 0
+
+ check_overlap = flow_add.flags & ofp.OFPFF_CHECK_OVERLAP
+ if check_overlap:
+ if self._flow_find_overlapping_flows(flow_add, return_on_first_hit=True):
+ self.signal_flow_mod_error(ofp.OFPFMFC_OVERLAP, flow_add)
+ else:
+ # free to add as new flow
+ flow = flow_stats_entry_from_flow_mod_message(flow_add)
+ self.flows.append(flow)
+
+ else:
+ flow = flow_stats_entry_from_flow_mod_message(flow_add)
+ idx = self._flow_find(flow)
+ if idx >= 0:
+ old_flow = self.flows[idx]
+ assert isinstance(old_flow, ofp.common.flow_stats_entry)
+ if not (flow_add.flags & ofp.OFPFF_RESET_COUNTS):
+ flow.byte_count = old_flow.byte_count
+ flow.packet_count = old_flow.packet_count
+ self.flows[idx] = flow
+
+ else:
+ self.flows.append(flow)
+
+ def flow_delete_strict(self, flow_delete_strict):
+ assert isinstance(flow_delete_strict, ofp.message.flow_delete_strict)
+ flow = flow_stats_entry_from_flow_mod_message(flow_delete_strict)
+ idx = self._flow_find(flow)
+ if (idx >= 0):
+ del self.flows[idx]
+ logging.info("flow removed:\n%s" % pp(flow))
+ else:
+ logging.error('Requesting strict delete of:\n%s\nwhen flow table is:\n\n' % pp(flow))
+ for f in self.flows:
+ print pp(f)
+
+ def flow_delete(self, flow_delete):
+ assert isinstance(flow_delete, ofp.message.flow_delete)
+
+ # build a list of what to keep vs what to delete
+ to_keep = []
+ to_delete = []
+ for f in self.flows:
+ list_to_append = to_delete if self._flow_matches_spec(f, flow_delete) else to_keep
+ list_to_append.append(f)
+
+ # replace flow table with keepers
+ self.flows = to_keep
+
+ # send notifications for discarded flows as required by OF spec
+ self._announce_flows_deleted(to_delete)
+
+ def flow_modify_strict(self, flow_obj):
+ raise Exception("flow_modify_strict(): Not implemented yet")
+
+ def flow_modify(self, flow_obj):
+ raise Exception("flow_modify(): Not implemented yet")
+
+ def flow_list(self):
+ return self.flows
+
+ def _flow_find_overlapping_flows(self, flow_mod, return_on_first_hit=False):
+ """
+ Return list of overlapping flow(s)
+ Two flows overlap if a packet may match both and if they have the same priority.
+ """
+
+ def _flow_find(self, flow):
+ for i, f in enumerate(self.flows):
+ if self._flows_match(f, flow):
+ return i
+ return -1
+
+ def _flows_match(self, f1, f2):
+ keys_matter = ('table_id', 'priority', 'flags', 'cookie', 'match')
+ for key in keys_matter:
+ if getattr(f1, key) != getattr(f2, key):
+ return False
+ return True
+
+ def _flow_matches_spec(self, flow, flow_mod):
+ """
+ Return True if a given flow (flow_stats_entry) is "covered" by the wildcarded flow
+ mod or delete spec (as defined in the flow_mod or flow_delete message); otherwise
+ return False.
+ """
+ #import pdb
+ #pdb.set_trace()
+
+ assert isinstance(flow, ofp.common.flow_stats_entry)
+ assert isinstance(flow_mod, (ofp.message.flow_delete, ofp.message.flow_mod))
+
+ # Check if flow.cookie is a match for flow_mod cookie/cookie_mask
+ if (flow.cookie & flow_mod.cookie_mask) != (flow_mod.cookie & flow_mod.cookie_mask):
+ return False
+
+ # Check if flow.table_id is covered by flow_mod.table_id
+ if flow_mod.table_id != ofp.OFPTT_ALL and flow.table_id != flow_mod.table_id:
+ return False
+
+ # Check out_port
+ if flow_mod.out_port != ofp.OFPP_ANY and not self._flow_has_out_port(flow, flow_mod.out_port):
+ return False
+
+ # Check out_group
+ if flow_mod.out_group != ofp.OFPG_ANY and not self._flow_has_out_group(flow, flow_mod.out_group):
+ return False
+
+ # Priority is ignored
+
+ # Check match condition
+ # If the flow_mod match field is empty, that is a special case and indicate the flow entry matches
+ match = flow_mod.match
+ assert isinstance(match, ofp.common.match_v3)
+ if not match.oxm_list:
+ return True # If we got this far and the match is empty in the flow spec, than the flow matches
+ else:
+ raise NotImplementedError("_flow_matches_spec(): No flow match analysys yet")
+
+ def _flow_has_out_port(self, flow, out_port):
+ """Return True if flow has a output command with the given out_port"""
+ assert isinstance(flow, ofp.common.flow_stats_entry)
+ for instruction in flow.instructions:
+ assert isinstance(instruction, ofp.instruction.instruction)
+ if instruction.type == ofp.OFPIT_APPLY_ACTIONS:
+ assert isinstance(instruction, ofp.instruction.apply_actions)
+ for action in instruction.actions:
+ assert isinstance(action, ofp.action.action)
+ if action.type == ofp.OFPAT_OUTPUT:
+ assert isinstance(action, ofp.action.output)
+ if action.port == out_port:
+ return True
+
+ # otherwise...
+ return False
+
+ def _flow_has_out_group(self, flow, out_group):
+ """Return True if flow has a output command with the given out_group"""
+ assert isinstance(flow, ofp.common.flow_stats_entry)
+ for instruction in flow.instructions:
+ assert isinstance(instruction, ofp.instruction.instruction)
+ if instruction.type == ofp.OFPIT_APPLY_ACTIONS:
+ assert isinstance(instruction, ofp.instruction.apply_actions)
+ for action in instruction.actions:
+ assert isinstance(action, ofp.action.action)
+ if action.type == ofp.OFPAT_GROUP:
+ assert isinstance(action, ofp.action.group)
+ if action.group_id == out_group:
+ return True
+
+ # otherwise...
+ return False
+
+ def _flows_delete_by_group_id(self, group_id):
+ """Delete any flow referring to given group id"""
+ to_keep = []
+ to_delete = []
+ for f in self.flows:
+ list_to_append = to_delete if self._flow_has_out_group(f, group_id) else to_keep
+ list_to_append.append(f)
+
+ # replace flow table with keepers
+ self.flows = to_keep
+
+ # send notifications for discarded flows as required by OF spec
+ self._announce_flows_deleted(to_delete)
+
+ def _announce_flows_deleted(self, flows):
+ for f in flows:
+ self._announce_flow_deleted(f)
+
+ def _announce_flow_deleted(self, flow):
+ """Send notification to controller if flow is flagged with OFPFF_SEND_FLOW_REM flag"""
+ if flow.flags & ofp.OFPFF_SEND_FLOW_REM:
+ raise NotImplementedError("_announce_flow_deleted()")
+
+
+ ## <=========================== GROUP HANDLERS =================================>
+
+ def group_add(self, group_add):
+ assert isinstance(group_add, ofp.message.group_add)
+ if group_add.group_id in self.groups:
+ self.signal_group_mod_error(ofp.OFPGMFC_GROUP_EXISTS, group_add)
+ else:
+ group_entry = group_entry_from_group_mod_message(group_add)
+ self.groups[group_add.group_id] = group_entry
+
+ def group_modify(self, group_modify):
+ assert isinstance(group_modify, ofp.message.group_modify)
+ if group_modify.group_id not in self.groups:
+ self.signal_group_mod_error(ofp.OFPGMFC_INVALID_GROUP, group_modify)
+ else:
+ # replace existing group entry with new group definition
+ group_entry = group_entry_from_group_mod_message(group_modify)
+ self.groups[group_modify.group_id] = group_entry
+
+ def group_delete(self, group_delete):
+ assert isinstance(group_delete, ofp.message.group_mod)
+ if group_delete.group_id == ofp.OFPG_ALL: # drop all groups
+ # we must delete all flows that point to this group and signal controller as
+ # requested by the flows' flag
+
+ self.groups = {}
+ logging.info("all groups deleted")
+
+ else:
+ if group_delete.group_id not in self.groups:
+ # per the spec, this is silent;y ignored
+ return
+
+ else:
+ self._flows_delete_by_group_id(group_delete.group_id)
+ del self.groups[group_delete.group_id]
+ logging.info("group %d deleted" % group_delete.group_id)
+
+ def group_list(self):
+ return [group_entry.group_desc for group_entry in self.groups.values()]
+
+ def group_stats(self):
+ return [group_entry.group_stats for group_entry in self.groups.values()]
+
+ ## <=========================== TABLE HANDLERS =================================>
+
+ def table_stats(self):
+ """Scan through flow entries and create table stats"""
+ stats = {}
+ for flow in self.flows:
+ table_id = flow.table_id
+ entry = stats.setdefault(table_id, ofp.common.table_stats_entry(table_id))
+ entry.active_count += 1
+ entry.lookup_count += 1 # FIXME how to count these?
+ entry.matched_count += 1 # FIXME how to count these?
+ stats[table_id] = entry
+ return sorted(stats.values(), key=lambda e: e.table_id)