Gitiles
Code Review Sign In
gerrit.opencord.org / igmpproxy / fcb3ccf720a9af7c8f747d248f38e55fac5295a2 / . / src / main / java / org / opencord / igmpproxy / IgmpManager.java
blob: d57c42e2ac96c6009ebdde4a8aee31ecc6e6b99d [file] [log] [blame]
David K. Bainbridged77028f2017-08-01 12:47:55 -0700[diff] [blame]1/*
Brian O'Connor4d084702017-08-03 22:45:58 -0700[diff] [blame]2 * Copyright 2017-present Open Networking Foundation
David K. Bainbridged77028f2017-08-01 12:47:55 -0700[diff] [blame]3 *
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 */
ke han81a38b92017-03-10 18:41:44 +0800[diff] [blame]16package org.opencord.igmpproxy;
17
18import com.google.common.collect.Maps;
19import org.apache.felix.scr.annotations.Activate;
20import org.apache.felix.scr.annotations.Component;
21import org.apache.felix.scr.annotations.Deactivate;
22import org.apache.felix.scr.annotations.Reference;
23import org.apache.felix.scr.annotations.ReferenceCardinality;
24import org.onlab.packet.EthType;
25import org.onlab.packet.Ethernet;
26import org.onlab.packet.IGMP;
27import org.onlab.packet.IGMPGroup;
28import org.onlab.packet.IGMPMembership;
29import org.onlab.packet.IGMPQuery;
30import org.onlab.packet.IPv4;
31import org.onlab.packet.Ip4Address;
32import org.onlab.packet.IpAddress;
33import org.onlab.packet.VlanId;
34import org.onosproject.core.ApplicationId;
35import org.onosproject.core.CoreService;
36import org.onosproject.incubator.net.config.basics.McastConfig;
37import org.onosproject.mastership.MastershipService;
38import org.onosproject.net.AnnotationKeys;
39import org.onosproject.net.ConnectPoint;
40import org.onosproject.net.DeviceId;
41import org.onosproject.net.Port;
42import org.onosproject.net.PortNumber;
43import org.onosproject.net.config.ConfigFactory;
44import org.onosproject.net.config.NetworkConfigEvent;
45import org.onosproject.net.config.NetworkConfigListener;
46import org.onosproject.net.config.NetworkConfigRegistry;
47import org.onosproject.net.config.basics.SubjectFactories;
48import org.onosproject.net.device.DeviceEvent;
49import org.onosproject.net.device.DeviceListener;
50import org.onosproject.net.device.DeviceService;
51import org.onosproject.net.flow.DefaultTrafficTreatment;
52import org.onosproject.net.flow.FlowRuleService;
53import org.onosproject.net.flow.criteria.Criteria;
54import org.onosproject.net.flowobjective.DefaultFilteringObjective;
55import org.onosproject.net.flowobjective.FilteringObjective;
56import org.onosproject.net.flowobjective.FlowObjectiveService;
57import org.onosproject.net.flowobjective.Objective;
58import org.onosproject.net.flowobjective.ObjectiveContext;
59import org.onosproject.net.flowobjective.ObjectiveError;
60import org.onosproject.net.mcast.McastRoute;
61import org.onosproject.net.mcast.MulticastRouteService;
62import org.onosproject.net.packet.InboundPacket;
63import org.onosproject.net.packet.PacketContext;
64import org.onosproject.net.packet.PacketProcessor;
65import org.onosproject.net.packet.PacketService;
66import org.opencord.cordconfig.access.AccessDeviceConfig;
67import org.opencord.cordconfig.access.AccessDeviceData;
68import org.slf4j.Logger;
69import org.slf4j.LoggerFactory;
70
71import java.util.ArrayList;
72import java.util.Collection;
73import java.util.Iterator;
74import java.util.List;
75import java.util.Map;
76import java.util.Set;
77import java.util.TimerTask;
78import java.util.concurrent.ConcurrentHashMap;
79import java.util.concurrent.Executors;
80import java.util.concurrent.ScheduledExecutorService;
81import java.util.concurrent.TimeUnit;
82
83/**
84 * Igmp process application, use proxy mode, support first join/ last leave , fast leave
85 * period query and keep alive, packet out igmp message to uplink port features.
86 */
87@Component(immediate = true)
88public class IgmpManager {
89
90 private static final Class<AccessDeviceConfig> CONFIG_CLASS =
91 AccessDeviceConfig.class;
92 private static final Class<IgmpproxyConfig> IGMPPROXY_CONFIG_CLASS =
93 IgmpproxyConfig.class;
94 private static final Class<IgmpproxySsmTranslateConfig> IGMPPROXY_SSM_CONFIG_CLASS =
95 IgmpproxySsmTranslateConfig.class;
96 private static final Class<McastConfig> MCAST_CONFIG_CLASS =
97 McastConfig.class;
98 public static Map<String, GroupMember> groupMemberMap = Maps.newConcurrentMap();
99 private static ApplicationId appId;
100 private static Map<DeviceId, AccessDeviceData> oltData = new ConcurrentHashMap<>();
101 private static int unSolicitedTimeout = 3; // unit is 1 sec
102 private static int keepAliveCount = 3;
103 private static int lastQueryInterval = 2; //unit is 1 sec
104 private static int lastQueryCount = 2;
105 private static boolean fastLeave = true;
106 private static boolean withRAUplink = true;
107 private static boolean withRADownlink = false;
108 private static boolean periodicQuery = true;
109 private static short mvlan = 4000;
110 private static byte igmpCos = 7;
111 public static boolean connectPointMode = true;
112 public static ConnectPoint connectPoint = null;
113
114 private final ScheduledExecutorService scheduledExecutorService =
115 Executors.newScheduledThreadPool(1);
116 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
117 protected CoreService coreService;
118 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
119 protected PacketService packetService;
120 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
121 protected MastershipService mastershipService;
122 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
123 protected FlowRuleService flowRuleService;
124 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
125 protected DeviceService deviceService;
126 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
127 protected FlowObjectiveService flowObjectiveService;
128 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
129 protected NetworkConfigRegistry networkConfig;
130 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
131 protected MulticastRouteService multicastService;
132 private IgmpPacketProcessor processor = new IgmpPacketProcessor();
133 private Logger log = LoggerFactory.getLogger(getClass());
134 private ApplicationId coreAppId;
135 private Map<Ip4Address, Ip4Address> ssmTranslateTable = new ConcurrentHashMap<>();
136 private InternalNetworkConfigListener configListener =
137 new InternalNetworkConfigListener();
138 private DeviceListener deviceListener = new InternalDeviceListener();
139 private ConfigFactory<DeviceId, AccessDeviceConfig> configFactory = null;
140 private ConfigFactory<ApplicationId, IgmpproxyConfig> igmpproxyConfigFactory =
141 new ConfigFactory<ApplicationId, IgmpproxyConfig>(
142 SubjectFactories.APP_SUBJECT_FACTORY, IGMPPROXY_CONFIG_CLASS, "igmpproxy") {
143 @Override
144 public IgmpproxyConfig createConfig() {
145 return new IgmpproxyConfig();
146 }
147 };
148 private ConfigFactory<ApplicationId, IgmpproxySsmTranslateConfig> igmpproxySsmConfigFactory =
149 new ConfigFactory<ApplicationId, IgmpproxySsmTranslateConfig>(
150 SubjectFactories.APP_SUBJECT_FACTORY, IGMPPROXY_SSM_CONFIG_CLASS, "ssmTranslate", true) {
151 @Override
152 public IgmpproxySsmTranslateConfig createConfig() {
153 return new IgmpproxySsmTranslateConfig();
154 }
155 };
156 private int maxResp = 10; //unit is 1 sec
157 private int keepAliveInterval = 120; //unit is 1 sec
158
159 public static int getUnsolicitedTimeout() {
160 return unSolicitedTimeout;
161 }
162
163 @Activate
164 protected void activate() {
165 appId = coreService.registerApplication("org.opencord.igmpproxy");
166 coreAppId = coreService.registerApplication(CoreService.CORE_APP_NAME);
167 packetService.addProcessor(processor, PacketProcessor.director(4));
168 IgmpSender.init(packetService, mastershipService);
169
170 if (networkConfig.getConfigFactory(CONFIG_CLASS) == null) {
171 configFactory =
172 new ConfigFactory<DeviceId, AccessDeviceConfig>(
173 SubjectFactories.DEVICE_SUBJECT_FACTORY, CONFIG_CLASS, "accessDevice") {
174 @Override
175 public AccessDeviceConfig createConfig() {
176 return new AccessDeviceConfig();
177 }
178 };
179 networkConfig.registerConfigFactory(configFactory);
180 }
181 networkConfig.registerConfigFactory(igmpproxySsmConfigFactory);
182 networkConfig.registerConfigFactory(igmpproxyConfigFactory);
183 networkConfig.addListener(configListener);
184
185 configListener.reconfigureNetwork(networkConfig.getConfig(appId, IGMPPROXY_CONFIG_CLASS));
186 configListener.reconfigureSsmTable(networkConfig.getConfig(appId, IGMPPROXY_SSM_CONFIG_CLASS));
187
188 networkConfig.getSubjects(DeviceId.class, AccessDeviceConfig.class).forEach(
189 subject -> {
190 AccessDeviceConfig config = networkConfig.getConfig(subject,
191 AccessDeviceConfig.class);
192 if (config != null) {
193 AccessDeviceData data = config.getAccessDevice();
194 oltData.put(data.deviceId(), data);
195 }
196 }
197 );
198
199 oltData.keySet().forEach(d->provisionDefaultFlows(d));
200 if (connectPointMode) {
201 provisionConnectPointFlows();
202 } else {
203 provisionUplinkFlows();
204 }
205
206 McastConfig config = networkConfig.getConfig(coreAppId, MCAST_CONFIG_CLASS);
207 if (config != null) {
208 mvlan = config.egressVlan().toShort();
209 }
210 deviceService.addListener(deviceListener);
211 scheduledExecutorService.scheduleAtFixedRate(new IgmpProxyTimerTask(), 0, 1000, TimeUnit.MILLISECONDS);
212
213 log.info("Started");
214 }
215
216 @Deactivate
217 protected void deactivate() {
218 scheduledExecutorService.shutdown();
219
220 // de-register and null our handler
221 networkConfig.removeListener(configListener);
222 if (configFactory != null) {
223 networkConfig.unregisterConfigFactory(configFactory);
224 }
225 networkConfig.unregisterConfigFactory(igmpproxyConfigFactory);
226 networkConfig.unregisterConfigFactory(igmpproxySsmConfigFactory);
227 deviceService.removeListener(deviceListener);
228 packetService.removeProcessor(processor);
229 flowRuleService.removeFlowRulesById(appId);
230
231 log.info("Stopped");
232 }
233
234 protected Ip4Address getDeviceIp(DeviceId ofDeviceId) {
235 try {
236 String[] mgmtAddress = deviceService.getDevice(ofDeviceId)
237 .annotations().value(AnnotationKeys.MANAGEMENT_ADDRESS).split(":");
238 return Ip4Address.valueOf(mgmtAddress[0]);
239 } catch (Exception ex) {
240 log.info("No valid Ipaddress for " + ofDeviceId.toString());
241 return null;
242 }
243 }
244
245 private void processIgmpQuery(IGMPQuery igmpQuery, ConnectPoint cp, int maxResp) {
246
247 DeviceId deviceId = cp.deviceId();
248 Ip4Address gAddr = igmpQuery.getGaddr().getIp4Address();
249
250 if (maxResp >= 128) {
251 int mant = maxResp & 0xf;
252 int exp = (maxResp >> 4) & 0x7;
253 maxResp = (mant | 0x10) << (exp + 3);
254 }
255
256 maxResp = (maxResp + 5) / 10;
257
258 if (gAddr != null && !gAddr.isZero()) {
259 StateMachine.specialQuery(deviceId, gAddr, maxResp);
260 } else {
261 StateMachine.generalQuery(deviceId, maxResp);
262 }
263
264 }
265
266 private Ip4Address ssmTranslateRoute(IpAddress group) {
267 return ssmTranslateTable.get(group);
268 }
269
270 private void processIgmpReport(IGMPMembership igmpGroup, VlanId vlan, ConnectPoint cp, byte igmpType) {
271 DeviceId deviceId = cp.deviceId();
272 PortNumber portNumber = cp.port();
273
274 Ip4Address groupIp = igmpGroup.getGaddr().getIp4Address();
275 if (!groupIp.isMulticast()) {
276 log.info(groupIp.toString() + " is not a valid group address");
277 return;
278 }
279 Ip4Address srcIp = getDeviceIp(deviceId);
280
281 byte recordType = igmpGroup.getRecordType();
282 boolean join = false;
283
284 ArrayList<Ip4Address> sourceList = new ArrayList<>();
285
286 if (igmpGroup.getSources().size() > 0) {
287 igmpGroup.getSources().forEach(source -> sourceList.add(source.getIp4Address()));
288 if (recordType == IGMPMembership.CHANGE_TO_EXCLUDE_MODE ||
289 recordType == IGMPMembership.MODE_IS_EXCLUDE ||
290 recordType == IGMPMembership.BLOCK_OLD_SOURCES) {
291 join = false;
292 } else if (recordType == IGMPMembership.CHANGE_TO_INCLUDE_MODE ||
293 recordType == IGMPMembership.MODE_IS_INCLUDE ||
294 recordType == IGMPMembership.ALLOW_NEW_SOURCES) {
295 join = true;
296 }
297 } else {
298 IpAddress src = ssmTranslateRoute(groupIp);
299 if (src == null) {
300 log.info("no ssm translate for group " + groupIp.toString());
301 return;
302 }
303 sourceList.add(src.getIp4Address());
304 if (recordType == IGMPMembership.CHANGE_TO_EXCLUDE_MODE ||
305 recordType == IGMPMembership.MODE_IS_EXCLUDE ||
306 recordType == IGMPMembership.BLOCK_OLD_SOURCES) {
307 join = true;
308 } else if (recordType == IGMPMembership.CHANGE_TO_INCLUDE_MODE ||
309 recordType == IGMPMembership.MODE_IS_INCLUDE ||
310 recordType == IGMPMembership.ALLOW_NEW_SOURCES) {
311 join = false;
312 }
313 }
314 String groupMemberKey = GroupMember.getkey(groupIp, deviceId, portNumber);
315 GroupMember groupMember = groupMemberMap.get(groupMemberKey);
316
317 if (join) {
318 if (groupMember == null) {
319 if (igmpType == IGMP.TYPE_IGMPV2_MEMBERSHIP_REPORT) {
320 groupMember = new GroupMember(groupIp, vlan, deviceId, portNumber, true);
321 } else {
322 groupMember = new GroupMember(groupIp, vlan, deviceId, portNumber, false);
323 }
324 StateMachine.join(deviceId, groupIp, srcIp);
325 groupMemberMap.put(groupMemberKey, groupMember);
326 groupMember.updateList(recordType, sourceList);
327 groupMember.getSourceList().forEach(source -> multicastService.addSink(
328 new McastRoute(source, groupIp, McastRoute.Type.IGMP), cp));
329 }
330 groupMember.resetAllTimers();
331 groupMember.updateList(recordType, sourceList);
332 groupMember.setLeave(false);
333 } else {
334 if (groupMember == null) {
335 log.info("receive leave but no instance, group " + groupIp.toString() +
336 " device:" + deviceId.toString() + " port:" + portNumber.toString());
337 return;
338 } else {
339 groupMember.setLeave(true);
340 if (fastLeave) {
341 leaveAction(groupMember);
342 } else {
343 sendQuery(groupMember);
344 }
345 }
346 }
347 }
348
349 private void leaveAction(GroupMember groupMember) {
350 ConnectPoint cp = new ConnectPoint(groupMember.getDeviceId(), groupMember.getPortNumber());
351 StateMachine.leave(groupMember.getDeviceId(), groupMember.getGroupIp());
352 groupMember.getSourceList().forEach(source -> multicastService.removeSink(
353 new McastRoute(source, groupMember.getGroupIp(),
354 McastRoute.Type.IGMP), cp));
355 groupMemberMap.remove(groupMember.getId());
356 }
357
358 private void sendQuery(GroupMember groupMember) {
359 Ethernet ethpkt;
360 Ip4Address srcIp = getDeviceIp(groupMember.getDeviceId());
361 if (groupMember.getv2()) {
362 ethpkt = IgmpSender.getInstance().buildIgmpV2Query(groupMember.getGroupIp(), srcIp);
363 } else {
364 ethpkt = IgmpSender.getInstance().buildIgmpV3Query(groupMember.getGroupIp(), srcIp);
365 }
366 IgmpSender.getInstance().sendIgmpPacket(ethpkt, groupMember.getDeviceId(), groupMember.getPortNumber());
367 }
368
369 private void processFilterObjective(DeviceId devId, Port port, boolean remove) {
370
371 //TODO migrate to packet requests when packet service uses filtering objectives
372 DefaultFilteringObjective.Builder builder = DefaultFilteringObjective.builder();
373
374 builder = remove ? builder.deny() : builder.permit();
375
376 FilteringObjective igmp = builder
377 .withKey(Criteria.matchInPort(port.number()))
378 .addCondition(Criteria.matchEthType(EthType.EtherType.IPV4.ethType()))
379 .addCondition(Criteria.matchIPProtocol(IPv4.PROTOCOL_IGMP))
380 .withMeta(DefaultTrafficTreatment.builder().setOutput(PortNumber.CONTROLLER).build())
381 .fromApp(appId)
382 .withPriority(10000)
383 .add(new ObjectiveContext() {
384 @Override
385 public void onSuccess(Objective objective) {
386 log.info("IgmpProxy filter for {} on {} installed.",
387 devId, port);
388 }
389
390 @Override
391 public void onError(Objective objective, ObjectiveError error) {
392 log.info("IgmpProxy filter for {} on {} failed because {}.",
393 devId, port, error);
394 }
395 });
396
397 flowObjectiveService.filter(devId, igmp);
398 }
399
400 /**
401 * Packet processor responsible for forwarding packets along their paths.
402 */
403 private class IgmpPacketProcessor implements PacketProcessor {
404 @Override
405 public void process(PacketContext context) {
406
407 try {
408 InboundPacket pkt = context.inPacket();
409 Ethernet ethPkt = pkt.parsed();
410 if (ethPkt == null) {
411 return;
412 }
413
414 if (ethPkt.getEtherType() != Ethernet.TYPE_IPV4) {
415 return;
416 }
417
418 IPv4 ipv4Pkt = (IPv4) ethPkt.getPayload();
419
420 if (ipv4Pkt.getProtocol() != IPv4.PROTOCOL_IGMP) {
421 return;
422 }
423
424 short vlan = ethPkt.getVlanID();
425 DeviceId deviceId = pkt.receivedFrom().deviceId();
426
427 if (oltData.get(deviceId) == null) {
428 log.error("Device not registered in netcfg :" + deviceId.toString());
429 return;
430 }
431
432 IGMP igmp = (IGMP) ipv4Pkt.getPayload();
433 switch (igmp.getIgmpType()) {
434 case IGMP.TYPE_IGMPV3_MEMBERSHIP_QUERY:
435 //Discard Query from OLT’s non-uplink port’s
436 if (!pkt.receivedFrom().port().equals(getDeviceUplink(deviceId))) {
437 log.info("IGMP Picked up query from non-uplink port");
438 return;
439 }
440
441 processIgmpQuery((IGMPQuery) igmp.getGroups().get(0), pkt.receivedFrom(),
442 0xff & igmp.getMaxRespField());
443 break;
444 case IGMP.TYPE_IGMPV1_MEMBERSHIP_REPORT:
445 log.debug("IGMP version 1 message types are not currently supported.");
446 break;
447 case IGMP.TYPE_IGMPV3_MEMBERSHIP_REPORT:
448 case IGMP.TYPE_IGMPV2_MEMBERSHIP_REPORT:
449 case IGMP.TYPE_IGMPV2_LEAVE_GROUP:
450 //Discard join/leave from OLT’s uplink port’s
451 if (pkt.receivedFrom().port().equals(getDeviceUplink(deviceId))) {
452 log.info("IGMP Picked up join/leave from the olt uplink port");
453 return;
454 }
455
456 Iterator<IGMPGroup> itr = igmp.getGroups().iterator();
457 while (itr.hasNext()) {
458 IGMPGroup group = itr.next();
459 if (group instanceof IGMPMembership) {
460 processIgmpReport((IGMPMembership) group, VlanId.vlanId(vlan),
461 pkt.receivedFrom(), igmp.getIgmpType());
462 } else if (group instanceof IGMPQuery) {
463 IGMPMembership mgroup;
464 mgroup = new IGMPMembership(group.getGaddr().getIp4Address());
465 mgroup.setRecordType(igmp.getIgmpType() == IGMP.TYPE_IGMPV2_MEMBERSHIP_REPORT ?
466 IGMPMembership.MODE_IS_EXCLUDE : IGMPMembership.MODE_IS_INCLUDE);
467 processIgmpReport(mgroup, VlanId.vlanId(vlan),
468 pkt.receivedFrom(), igmp.getIgmpType());
469 }
470 }
471 break;
472
473 default:
474 log.info("wrong IGMP v3 type:" + igmp.getIgmpType());
475 break;
476 }
477
478 } catch (Exception ex) {
479 log.error("igmp process error : " + ex.toString());
480 ex.printStackTrace();
481 }
482 }
483 }
484
485 private class IgmpProxyTimerTask extends TimerTask {
486 public void run() {
487 try {
488 IgmpTimer.timeOut1s();
489 queryMembers();
490 } catch (Exception ex) {
491 log.warn("Igmp timer task error : {}", ex.getMessage());
492 }
493 }
494
495 private void queryMembers() {
496 GroupMember groupMember;
497 Set groupMemberSet = groupMemberMap.entrySet();
498 Iterator itr = groupMemberSet.iterator();
499 while (itr.hasNext()) {
500 Map.Entry entry = (Map.Entry) itr.next();
501 groupMember = (GroupMember) entry.getValue();
502 DeviceId did = groupMember.getDeviceId();
503 if (mastershipService.isLocalMaster(did)) {
504 if (groupMember.isLeave()) {
505 lastQuery(groupMember);
506 } else if (periodicQuery) {
507 periodicQuery(groupMember);
508 }
509 }
510 }
511 }
512
513 private void lastQuery(GroupMember groupMember) {
514 if (groupMember.getLastQueryInterval() < lastQueryInterval) {
515 groupMember.lastQueryInterval(true); // count times
516 } else if (groupMember.getLastQueryCount() < lastQueryCount - 1) {
517 sendQuery(groupMember);
518 groupMember.lastQueryInterval(false); // reset count number
519 groupMember.lastQueryCount(true); //count times
520 } else if (groupMember.getLastQueryCount() == lastQueryCount - 1) {
521 leaveAction(groupMember);
522 }
523 }
524
525 private void periodicQuery(GroupMember groupMember) {
526 if (groupMember.getKeepAliveQueryInterval() < keepAliveInterval) {
527 groupMember.keepAliveInterval(true);
528 } else if (groupMember.getKeepAliveQueryCount() < keepAliveCount) {
529 sendQuery(groupMember);
530 groupMember.keepAliveInterval(false);
531 groupMember.keepAliveQueryCount(true);
532 } else if (groupMember.getKeepAliveQueryCount() == keepAliveCount) {
533 leaveAction(groupMember);
534 }
535 }
536
537 }
538
539 public static PortNumber getDeviceUplink(DeviceId devId) {
540 return oltData.get(devId).uplink();
541 }
542
543 private void processFilterObjective(DeviceId devId, PortNumber port, boolean remove) {
544 //TODO migrate to packet requests when packet service uses filtering objectives
545 DefaultFilteringObjective.Builder builder = DefaultFilteringObjective.builder();
546
547 builder = remove ? builder.deny() : builder.permit();
548
549 FilteringObjective igmp = builder
550 .withKey(Criteria.matchInPort(port))
551 .addCondition(Criteria.matchEthType(EthType.EtherType.IPV4.ethType()))
552 .addCondition(Criteria.matchIPProtocol(IPv4.PROTOCOL_IGMP))
553 .withMeta(DefaultTrafficTreatment.builder().setOutput(PortNumber.CONTROLLER).build())
554 .fromApp(appId)
555 .withPriority(10000)
556 .add(new ObjectiveContext() {
557 @Override
558 public void onSuccess(Objective objective) {
559 log.info("IgmpProxy filter for {} on {} installed.",
560 devId, port);
561 }
562
563 @Override
564 public void onError(Objective objective, ObjectiveError error) {
565 log.info("IgmpProxy filter for {} on {} failed because {}.",
566 devId, port, error);
567 }
568 });
569
570 flowObjectiveService.filter(devId, igmp);
571 }
572
573 private boolean isConnectPoint(DeviceId device, PortNumber port) {
Deepa Vaddireddy01f33b42017-07-02 13:26:53 +0530[diff] [blame]574 if (connectPoint != null) {
575 return (connectPointMode && connectPoint.deviceId().equals(device)
576 && connectPoint.port().equals(port));
577 } else {
578 log.info("connectPoint not configured for device {}", device);
579 return false;
580 }
ke han81a38b92017-03-10 18:41:44 +0800[diff] [blame]581 }
Deepa Vaddireddy01f33b42017-07-02 13:26:53 +0530[diff] [blame]582
ke han81a38b92017-03-10 18:41:44 +0800[diff] [blame]583 private boolean isUplink(DeviceId device, PortNumber port) {
584 return ((!connectPointMode) && oltData.containsKey(device)
585 && oltData.get(device).uplink().equals(port));
586 }
587
588 private class InternalDeviceListener implements DeviceListener {
589 @Override
590 public void event(DeviceEvent event) {
591 DeviceId devId = event.subject().id();
592 PortNumber port;
593 if (oltData.get(devId) == null) {
594 return;
595 }
596 switch (event.type()) {
597
598 case DEVICE_ADDED:
599 case DEVICE_UPDATED:
600 case DEVICE_REMOVED:
601 case DEVICE_SUSPENDED:
602 case DEVICE_AVAILABILITY_CHANGED:
603 case PORT_STATS_UPDATED:
604 break;
605 case PORT_ADDED:
606 port = event.port().number();
607 if (oltData.containsKey(devId) && !isUplink(devId, port) && !isConnectPoint(devId, port)) {
608 processFilterObjective(devId, port, false);
609 } else if (isUplink(devId, port)) {
610 provisionUplinkFlows();
611 } else if (isConnectPoint(devId, port)) {
612 provisionConnectPointFlows();
613 }
614 break;
615 case PORT_UPDATED:
616 port = event.port().number();
617 if (oltData.containsKey(devId) && !isUplink(devId, port) && !isConnectPoint(devId, port)) {
618 if (event.port().isEnabled()) {
619 processFilterObjective(devId, port, false);
620 } else {
621 processFilterObjective(devId, port, true);
622 }
623 } else if (isUplink(devId, port)) {
624 if (event.port().isEnabled()) {
625 provisionUplinkFlows(devId);
626 } else {
627 processFilterObjective(devId, port, true);
628 }
629 } else if (isConnectPoint(devId, port)) {
630 if (event.port().isEnabled()) {
631 provisionConnectPointFlows();
632 } else {
633 unprovisionConnectPointFlows();
634 }
635 }
636 break;
637 case PORT_REMOVED:
638 port = event.port().number();
639 processFilterObjective(devId, port, true);
640 break;
641 default:
642 log.info("Unknown device event {}", event.type());
643 break;
644 }
645 }
646
647 @Override
648 public boolean isRelevant(DeviceEvent event) {
649 return true;
650 }
651 }
652
653 private class InternalNetworkConfigListener implements NetworkConfigListener {
654
655 private void reconfigureNetwork(IgmpproxyConfig cfg) {
656 IgmpproxyConfig newCfg;
657 if (cfg == null) {
658 newCfg = new IgmpproxyConfig();
659 } else {
660 newCfg = cfg;
661 }
662
663 unSolicitedTimeout = newCfg.unsolicitedTimeOut();
664 maxResp = newCfg.maxResp();
665 keepAliveInterval = newCfg.keepAliveInterval();
666 keepAliveCount = newCfg.keepAliveCount();
667 lastQueryInterval = newCfg.lastQueryInterval();
668 lastQueryCount = newCfg.lastQueryCount();
669 withRAUplink = newCfg.withRAUplink();
670 withRADownlink = newCfg.withRADownlink();
671 igmpCos = newCfg.igmpCos();
672 periodicQuery = newCfg.periodicQuery();
673 fastLeave = newCfg.fastLeave();
674
675 connectPoint = newCfg.connectPoint();
676 if (connectPointMode != newCfg.connectPointMode()) {
677 connectPointMode = newCfg.connectPointMode();
678 if (connectPointMode) {
679 unprovisionUplinkFlows();
680 provisionConnectPointFlows();
681 } else {
682 unprovisionConnectPointFlows();
683 provisionUplinkFlows();
684 }
685 }
686 if (connectPoint != null) {
687 log.info("connect point :" + connectPoint.toString());
688 }
689 log.info(" mode: " + connectPointMode);
690
691 IgmpSender.getInstance().setIgmpCos(igmpCos);
692 IgmpSender.getInstance().setMaxResp(maxResp);
693 IgmpSender.getInstance().setMvlan(mvlan);
694 IgmpSender.getInstance().setWithRADownlink(withRADownlink);
695 IgmpSender.getInstance().setWithRAUplink(withRAUplink);
696
697 }
698
699 public void reconfigureSsmTable(IgmpproxySsmTranslateConfig cfg) {
700 if (cfg == null) {
701 return;
702 }
703 Collection<McastRoute> translations = cfg.getSsmTranslations();
704 for (McastRoute route : translations) {
705 ssmTranslateTable.put(route.group().getIp4Address(), route.source().getIp4Address());
706 }
707 }
708
709 @Override
710 public void event(NetworkConfigEvent event) {
711 switch (event.type()) {
712 case CONFIG_ADDED:
713 case CONFIG_UPDATED:
714 if (event.configClass().equals(CONFIG_CLASS)) {
715 AccessDeviceConfig config =
716 networkConfig.getConfig((DeviceId) event.subject(), CONFIG_CLASS);
717 if (config != null) {
718 oltData.put(config.getAccessDevice().deviceId(), config.getAccessDevice());
719 provisionDefaultFlows((DeviceId) event.subject());
720 provisionUplinkFlows((DeviceId) event.subject());
721 }
722 }
723
724 if (event.configClass().equals(IGMPPROXY_CONFIG_CLASS)) {
725 IgmpproxyConfig config = networkConfig.getConfig(appId, IGMPPROXY_CONFIG_CLASS);
726 if (config != null) {
727 reconfigureNetwork(config);
728 }
729 }
730
731 if (event.configClass().equals(IGMPPROXY_SSM_CONFIG_CLASS)) {
732 IgmpproxySsmTranslateConfig config = networkConfig.getConfig(appId, IGMPPROXY_SSM_CONFIG_CLASS);
733 if (config != null) {
734 reconfigureSsmTable(config);
735 }
736 }
737
738 if (event.configClass().equals(MCAST_CONFIG_CLASS)) {
739 McastConfig config = networkConfig.getConfig(coreAppId, MCAST_CONFIG_CLASS);
740 if (config != null && mvlan != config.egressVlan().toShort()) {
741 mvlan = config.egressVlan().toShort();
742 groupMemberMap.values().forEach(m -> leaveAction(m));
743 }
744 }
745
746 log.info("Reconfigured");
747 break;
748 case CONFIG_REGISTERED:
749 case CONFIG_UNREGISTERED:
750 break;
751 case CONFIG_REMOVED:
752 if (event.configClass().equals(CONFIG_CLASS)) {
753 oltData.remove(event.subject());
754 }
755
756 default:
757 break;
758 }
759 }
760 }
761
762 private void provisionDefaultFlows(DeviceId deviceId) {
763 List<Port> ports = deviceService.getPorts(deviceId);
764 ports.stream()
765 .filter(p -> (!oltData.get(p.element().id()).uplink().equals(p.number()) && p.isEnabled()))
766 .forEach(p -> processFilterObjective((DeviceId) p.element().id(), p.number(), false));
767 }
768
769 private void provisionUplinkFlows(DeviceId deviceId) {
770 if (connectPointMode) {
771 return;
772 }
773
774 processFilterObjective(deviceId, oltData.get(deviceId).uplink(), false);
775 }
776
777 private void provisionUplinkFlows() {
778 if (connectPointMode) {
779 return;
780 }
781
782 oltData.keySet().forEach(deviceId ->provisionUplinkFlows(deviceId));
783 }
784 private void unprovisionUplinkFlows() {
785 oltData.keySet().forEach(deviceId ->
786 processFilterObjective(deviceId, oltData.get(deviceId).uplink(), true));
787 }
788
789 private void provisionConnectPointFlows() {
790 if ((!connectPointMode) || connectPoint == null) {
791 return;
792 }
793
794 processFilterObjective(connectPoint.deviceId(), connectPoint.port(), false);
795 }
796 private void unprovisionConnectPointFlows() {
797 if (connectPoint == null) {
798 return;
799 }
800 processFilterObjective(connectPoint.deviceId(), connectPoint.port(), true);
801 }
802}