blob: 6117c8fd8326416ce8dd8c19eeacaba8b35fcd94 [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
Girishf6eeaea2017-11-13 10:53:57 +053020import os.path
Jonathan Hart8d21c322018-04-17 07:42:02 -070021from twisted.internet import protocol, reactor, ssl
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070022from 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
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070028
29log = structlog.get_logger()
30
31
32class Agent(protocol.ClientFactory):
33
sathishg00433e52017-05-23 16:07:41 +053034 generation_is_defined = False
35 cached_generation_id = None
36
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070037 def __init__(self,
38 controller_endpoint,
39 datapath_id,
40 device_id,
41 rpc_stub,
Girishf6eeaea2017-11-13 10:53:57 +053042 enable_tls=False,
43 key_file=None,
44 cert_file=None,
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070045 conn_retry_interval=1):
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070046
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070047 self.controller_endpoint = controller_endpoint
48 self.datapath_id = datapath_id
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070049 self.device_id = device_id
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070050 self.rpc_stub = rpc_stub
Girishf6eeaea2017-11-13 10:53:57 +053051 self.enable_tls = enable_tls
52 self.key_file = key_file
53 self.cert_file = cert_file
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070054 self.retry_interval = conn_retry_interval
55
56 self.running = False
57 self.connector = None # will be a Connector instance once connected
58 self.d_disconnected = None # a deferred to signal reconnect loop when
59 # TCP connection is lost
60 self.connected = False
61 self.exiting = False
Zsolt Haraszti47324632016-12-20 00:50:33 -080062 self.proto_handler = None
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070063
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070064 def get_device_id(self):
65 return self.device_id
66
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070067 def start(self):
68 log.debug('starting')
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070069 if self.running:
70 return
71 self.running = True
72 reactor.callLater(0, self.keep_connected)
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070073 log.info('started')
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070074 return self
75
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070076 def stop(self):
77 log.debug('stopping')
78 self.connected = False
79 self.exiting = True
80 self.connector.disconnect()
81 log.info('stopped')
82
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070083 def resolve_endpoint(self, endpoint):
alshabibc3fb4942017-01-26 15:34:24 -080084 # enable optional resolution via consul;
85 # see https://jira.opencord.org/browse/CORD-820
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070086 host, port = endpoint.split(':', 2)
87 return host, int(port)
88
89 @inlineCallbacks
90 def keep_connected(self):
91 """Keep reconnecting to the controller"""
92 while not self.exiting:
93 host, port = self.resolve_endpoint(self.controller_endpoint)
94 log.info('connecting', host=host, port=port)
Girishf6eeaea2017-11-13 10:53:57 +053095 if self.enable_tls:
96 try:
97 # Check that key_file and cert_file is provided and
98 # the files exist
99 if self.key_file is None or \
100 self.cert_file is None or \
101 not os.path.isfile(self.key_file) or \
102 not os.path.isfile(self.cert_file):
103 raise Exception('key_file "{}" or cert_file "{}"'
104 ' is not found'.
105 format(self.key_file, self.cert_file))
106 with open(self.key_file) as keyFile:
107 with open(self.cert_file) as certFile:
108 clientCert = ssl.PrivateCertificate.loadPEM(
109 keyFile.read() + certFile.read())
schowdhury3fbd4702017-06-30 03:50:25 -0700110
Girishf6eeaea2017-11-13 10:53:57 +0530111 ctx = clientCert.options()
112 self.connector = reactor.connectSSL(host, port, self, ctx)
113 log.info('tls-enabled')
schowdhury3fbd4702017-06-30 03:50:25 -0700114
Girishf6eeaea2017-11-13 10:53:57 +0530115 except Exception as e:
116 log.exception('failed-to-connect', reason=e)
117 else:
118 self.connector = reactor.connectTCP(host, port, self)
119 log.info('tls-disabled')
schowdhury3fbd4702017-06-30 03:50:25 -0700120
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700121 self.d_disconnected = Deferred()
122 yield self.d_disconnected
123 log.debug('reconnect', after_delay=self.retry_interval)
124 yield asleep(self.retry_interval)
125
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700126 def enter_disconnected(self, event, reason):
127 """Internally signal entering disconnected state"""
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700128 self.connected = False
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700129 if not self.exiting:
130 log.error(event, reason=reason)
131 self.d_disconnected.callback(None)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700132
133 def enter_connected(self):
134 """Handle transitioning from disconnected to connected state"""
135 log.info('connected')
136 self.connected = True
137 self.read_buffer = None
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700138 reactor.callLater(0, self.proto_handler.start)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700139
140 # protocol.ClientFactory methods
141
142 def protocol(self):
143 cxn = OpenFlowConnection(self) # Low level message handler
144 self.proto_handler = OpenFlowProtocolHandler(
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700145 self.datapath_id, self.device_id, self, cxn, self.rpc_stub)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700146 return cxn
147
148 def clientConnectionFailed(self, connector, reason):
149 self.enter_disconnected('connection-failed', reason)
150
151 def clientConnectionLost(self, connector, reason):
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700152 if not self.exiting:
153 log.error('client-connection-lost',
154 reason=reason, connector=connector)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700155
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700156 def forward_packet_in(self, ofp_packet_in):
Zsolt Haraszti47324632016-12-20 00:50:33 -0800157 if self.proto_handler is not None:
158 self.proto_handler.forward_packet_in(ofp_packet_in)
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700159
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800160 def forward_change_event(self, event):
161 # assert isinstance(event, ChangeEvent)
162 log.info('got-change-event', change_event=event)
163 if event.HasField("port_status"):
Zsolt Haraszti47324632016-12-20 00:50:33 -0800164 if self.proto_handler is not None:
165 self.proto_handler.forward_port_status(event.port_status)
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800166 else:
167 log.error('unknown-change-event', change_event=event)
168
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700169
170if __name__ == '__main__':
171 """Run this to test the agent for N concurrent sessions:
172 python agent [<number-of-desired-instances>]
173 """
174
175 n = 1 if len(sys.argv) < 2 else int(sys.argv[1])
176
177 from utils import mac_str_to_tuple
178
179 class MockRpc(object):
180 @staticmethod
181 def get_port_list(_):
182 ports = []
183 cap = of13.OFPPF_1GB_FD | of13.OFPPF_FIBER
184 for pno, mac, nam, cur, adv, sup, spe in (
185 (1, '00:00:00:00:00:01', 'onu1', cap, cap, cap,
186 of13.OFPPF_1GB_FD),
187 (2, '00:00:00:00:00:02', 'onu2', cap, cap, cap,
188 of13.OFPPF_1GB_FD),
189 (129, '00:00:00:00:00:81', 'olt', cap, cap, cap,
190 of13.OFPPF_1GB_FD)
191 ):
192 port = of13.common.port_desc(pno, mac_str_to_tuple(mac), nam,
193 curr=cur, advertised=adv,
194 supported=sup,
195 curr_speed=spe, max_speed=spe)
196 ports.append(port)
197 return ports
198
199 stub = MockRpc()
Zsolt Harasztiee5c4c82017-01-09 14:37:57 -0800200 agents = [Agent('localhost:6653', 256 + i, stub).start() for i in range(n)]
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700201
202 def shutdown():
203 [a.stop() for a in agents]
204
205 reactor.addSystemEventTrigger('before', 'shutdown', shutdown)
206 reactor.run()