blob: a222480d549cc5bcc540a1b900dd0f61146d820b [file] [log] [blame]
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -07001#
Zsolt Haraszti3eb27a52017-01-03 21:56:48 -08002# Copyright 2017 the original author or authors.
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -07003#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17import sys
18
19import structlog
20from twisted.internet import protocol
21from twisted.internet import reactor
22from twisted.internet.defer import Deferred, inlineCallbacks
23
24import loxi.of13 as of13
25from common.utils.asleep import asleep
26from of_connection import OpenFlowConnection
27from of_protocol_handler import OpenFlowProtocolHandler
Khen Nursimulu7626ce12016-12-21 11:51:46 -050028# from ofagent.protos.openflow_13_pb2 import ChangeEvent
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070029
30log = structlog.get_logger()
31
32
33class Agent(protocol.ClientFactory):
34
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070035 def __init__(self,
36 controller_endpoint,
37 datapath_id,
38 device_id,
39 rpc_stub,
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070040 conn_retry_interval=1):
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070041
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070042 self.controller_endpoint = controller_endpoint
43 self.datapath_id = datapath_id
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070044 self.device_id = device_id
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070045 self.rpc_stub = rpc_stub
46 self.retry_interval = conn_retry_interval
47
48 self.running = False
49 self.connector = None # will be a Connector instance once connected
50 self.d_disconnected = None # a deferred to signal reconnect loop when
51 # TCP connection is lost
52 self.connected = False
53 self.exiting = False
Zsolt Haraszti47324632016-12-20 00:50:33 -080054 self.proto_handler = None
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070055
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070056 def get_device_id(self):
57 return self.device_id
58
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070059 def start(self):
60 log.debug('starting')
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070061 if self.running:
62 return
63 self.running = True
64 reactor.callLater(0, self.keep_connected)
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070065 log.info('started')
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070066 return self
67
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070068 def stop(self):
69 log.debug('stopping')
70 self.connected = False
71 self.exiting = True
72 self.connector.disconnect()
73 log.info('stopped')
74
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070075 def resolve_endpoint(self, endpoint):
76 # TODO allow resolution via consul
77 host, port = endpoint.split(':', 2)
78 return host, int(port)
79
80 @inlineCallbacks
81 def keep_connected(self):
82 """Keep reconnecting to the controller"""
83 while not self.exiting:
84 host, port = self.resolve_endpoint(self.controller_endpoint)
85 log.info('connecting', host=host, port=port)
86 self.connector = reactor.connectTCP(host, port, self)
87 self.d_disconnected = Deferred()
88 yield self.d_disconnected
89 log.debug('reconnect', after_delay=self.retry_interval)
90 yield asleep(self.retry_interval)
91
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070092 def enter_disconnected(self, event, reason):
93 """Internally signal entering disconnected state"""
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070094 self.connected = False
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070095 if not self.exiting:
96 log.error(event, reason=reason)
97 self.d_disconnected.callback(None)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070098
99 def enter_connected(self):
100 """Handle transitioning from disconnected to connected state"""
101 log.info('connected')
102 self.connected = True
103 self.read_buffer = None
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700104 reactor.callLater(0, self.proto_handler.start)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700105
106 # protocol.ClientFactory methods
107
108 def protocol(self):
109 cxn = OpenFlowConnection(self) # Low level message handler
110 self.proto_handler = OpenFlowProtocolHandler(
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700111 self.datapath_id, self.device_id, self, cxn, self.rpc_stub)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700112 return cxn
113
114 def clientConnectionFailed(self, connector, reason):
115 self.enter_disconnected('connection-failed', reason)
116
117 def clientConnectionLost(self, connector, reason):
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700118 if not self.exiting:
119 log.error('client-connection-lost',
120 reason=reason, connector=connector)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700121
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700122 def forward_packet_in(self, ofp_packet_in):
Zsolt Haraszti47324632016-12-20 00:50:33 -0800123 if self.proto_handler is not None:
124 self.proto_handler.forward_packet_in(ofp_packet_in)
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700125
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800126 def forward_change_event(self, event):
127 # assert isinstance(event, ChangeEvent)
128 log.info('got-change-event', change_event=event)
129 if event.HasField("port_status"):
Zsolt Haraszti47324632016-12-20 00:50:33 -0800130 if self.proto_handler is not None:
131 self.proto_handler.forward_port_status(event.port_status)
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800132 else:
133 log.error('unknown-change-event', change_event=event)
134
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700135
136if __name__ == '__main__':
137 """Run this to test the agent for N concurrent sessions:
138 python agent [<number-of-desired-instances>]
139 """
140
141 n = 1 if len(sys.argv) < 2 else int(sys.argv[1])
142
143 from utils import mac_str_to_tuple
144
145 class MockRpc(object):
146 @staticmethod
147 def get_port_list(_):
148 ports = []
149 cap = of13.OFPPF_1GB_FD | of13.OFPPF_FIBER
150 for pno, mac, nam, cur, adv, sup, spe in (
151 (1, '00:00:00:00:00:01', 'onu1', cap, cap, cap,
152 of13.OFPPF_1GB_FD),
153 (2, '00:00:00:00:00:02', 'onu2', cap, cap, cap,
154 of13.OFPPF_1GB_FD),
155 (129, '00:00:00:00:00:81', 'olt', cap, cap, cap,
156 of13.OFPPF_1GB_FD)
157 ):
158 port = of13.common.port_desc(pno, mac_str_to_tuple(mac), nam,
159 curr=cur, advertised=adv,
160 supported=sup,
161 curr_speed=spe, max_speed=spe)
162 ports.append(port)
163 return ports
164
165 stub = MockRpc()
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700166 agents = [Agent('localhost:6633', 256 + i, stub).start() for i in range(n)]
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700167
168 def shutdown():
169 [a.stop() for a in agents]
170
171 reactor.addSystemEventTrigger('before', 'shutdown', shutdown)
172 reactor.run()
173