1 from vpp_object import VppObject
2 from vpp_ip import INVALID_INDEX
11 class PolicerAction():
12 """ sse2 qos action """
14 def __init__(self, type, dscp):
19 return {'type': self.type, 'dscp': self.dscp}
22 class VppPolicer(VppObject):
25 def __init__(self, test, name, cir, eir, commited_burst, excess_burst,
26 rate_type=0, round_type=0, type=0, color_aware=False,
27 conform_action=PolicerAction(1, 0),
28 exceed_action=PolicerAction(0, 0),
29 violate_action=PolicerAction(0, 0)):
34 self.commited_burst = commited_burst
35 self.excess_burst = excess_burst
36 self.rate_type = rate_type
37 self.round_type = round_type
39 self.color_aware = color_aware
40 self.conform_action = conform_action
41 self.exceed_action = exceed_action
42 self.violate_action = violate_action
43 self._policer_index = INVALID_INDEX
46 def policer_index(self):
47 return self._policer_index
49 def add_vpp_config(self):
50 r = self._test.vapi.policer_add_del(
51 name=self.name, cir=self.cir,
52 eir=self.eir, cb=self.commited_burst, eb=self.excess_burst,
53 rate_type=self.rate_type, round_type=self.round_type,
54 type=self.type, color_aware=self.color_aware,
55 conform_action=self.conform_action.encode(),
56 exceed_action=self.exceed_action.encode(),
57 violate_action=self.violate_action.encode())
58 self._test.registry.register(self, self._test.logger)
59 self._policer_index = r.policer_index
62 def remove_vpp_config(self):
63 self._test.vapi.policer_add_del(is_add=False, name=self.name)
64 self._policer_index = INVALID_INDEX
66 def bind_vpp_config(self, worker, bind):
67 self._test.vapi.policer_bind(name=self.name, worker_index=worker,
70 def apply_vpp_config(self, if_index, dir: Dir, apply):
72 self._test.vapi.policer_input(
73 name=self.name, sw_if_index=if_index, apply=apply)
75 self._test.vapi.policer_output(
76 name=self.name, sw_if_index=if_index, apply=apply)
78 def query_vpp_config(self):
79 dump = self._test.vapi.policer_dump(
80 match_name_valid=True, match_name=self.name)
82 if policer.name == self.name:
87 return ("policer-%s" % (self.name))
89 def get_stats(self, worker=None):
90 conform = self._test.statistics.get_counter("/net/policer/conform")
91 exceed = self._test.statistics.get_counter("/net/policer/exceed")
92 violate = self._test.statistics.get_counter("/net/policer/violate")
94 counters = {"conform": conform, "exceed": exceed, "violate": violate}
97 for name, c in counters.items():
98 total[f'{name}_packets'] = 0
99 total[f'{name}_bytes'] = 0
100 for i in range(len(c)):
102 if worker is not None and i != worker + 1:
104 stat_index = self._policer_index
105 total[f'{name}_packets'] += t[stat_index]['packets']
106 total[f'{name}_bytes'] += t[stat_index]['bytes']