affinity/implementation/src/main/java/org/opendaylight/affinity/affinity/internal/AffinityManagerImpl.java

   1 /*
   2  * Copyright (c) 2013 Plexxi, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.affinity.affinity.internal;
  10
  11 import java.io.FileNotFoundException;
  12 import java.io.IOException;
  13 import java.io.ObjectInputStream;
  14 import java.net.UnknownHostException;
  15 import java.net.InetAddress;
  16 import java.net.NetworkInterface;
  17 import java.net.SocketException;
  18 import java.util.ArrayList;
  19 import java.util.Collections;
  20 import java.util.Date;
  21 import java.util.Dictionary;
  22 import java.util.EnumSet;
  23 import java.util.Enumeration;
  24 import java.util.HashMap;
  25 import java.util.HashSet;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.AbstractMap;
  29 import java.util.Set;
  30 import java.util.concurrent.ConcurrentHashMap;
  31 import java.util.concurrent.ConcurrentMap;
  32 import java.util.concurrent.CopyOnWriteArrayList;
  33 import java.util.ArrayList;
  34 import java.util.EnumSet;
  35 import java.util.HashMap;
  36 import java.util.HashSet;
  37 import java.util.List;
  38 import java.util.Map;
  39 import java.util.Map.Entry;
  40 import java.util.Set;
  41 import java.util.concurrent.ConcurrentHashMap;
  42 import java.util.concurrent.ConcurrentMap;
  43
  44 import org.apache.felix.dm.Component;
  45 import org.eclipse.osgi.framework.console.CommandInterpreter;
  46 import org.eclipse.osgi.framework.console.CommandProvider;
  47 import org.opendaylight.controller.clustering.services.CacheConfigException;
  48 import org.opendaylight.controller.clustering.services.CacheExistException;
  49 import org.opendaylight.controller.clustering.services.ICacheUpdateAware;
  50 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
  51 import org.opendaylight.controller.clustering.services.IClusterServices;
  52 import org.opendaylight.controller.configuration.IConfigurationContainerAware;
  53 import org.opendaylight.controller.forwardingrulesmanager.FlowEntry;
  54 import org.opendaylight.controller.sal.core.IContainer;
  55 import org.opendaylight.controller.sal.core.Node;
  56 import org.opendaylight.controller.sal.core.Host;
  57 import org.opendaylight.controller.sal.core.NodeConnector;
  58 import org.opendaylight.controller.sal.core.NodeTable;
  59 import org.opendaylight.controller.sal.core.Property;
  60 import org.opendaylight.controller.sal.core.UpdateType;
  61 import org.opendaylight.controller.sal.flowprogrammer.IFlowProgrammerService;
  62 import org.opendaylight.controller.sal.flowprogrammer.Flow;
  63 import org.opendaylight.controller.sal.match.Match;
  64 import org.opendaylight.controller.sal.match.MatchType;
  65 import org.opendaylight.controller.sal.match.MatchField;
  66 import org.opendaylight.controller.sal.action.Action;
  67 import org.opendaylight.controller.sal.action.Output;
  68
  69 import org.opendaylight.controller.sal.reader.FlowOnNode;
  70 import org.opendaylight.controller.sal.reader.IReadService;
  71 import org.opendaylight.controller.sal.reader.IReadServiceListener;
  72 import org.opendaylight.controller.sal.utils.GlobalConstants;
  73 import org.opendaylight.controller.sal.utils.IObjectReader;
  74 import org.opendaylight.controller.sal.utils.ObjectReader;
  75 import org.opendaylight.controller.sal.utils.ObjectWriter;
  76 import org.opendaylight.controller.sal.utils.NetUtils;
  77
  78 import org.opendaylight.controller.hosttracker.IfIptoHost;
  79 import org.opendaylight.controller.sal.utils.Status;
  80 import org.opendaylight.controller.sal.utils.StatusCode;
  81
  82 import org.opendaylight.controller.sal.utils.ServiceHelper;
  83 import org.opendaylight.affinity.affinity.AffinityGroup;
  84 import org.opendaylight.affinity.affinity.AffinityLink;
  85 import org.opendaylight.affinity.affinity.AffinityIdentifier;
  86 import org.opendaylight.affinity.affinity.IAffinityManager;
  87 import org.opendaylight.affinity.affinity.IAffinityManagerAware;
  88
  89 import org.opendaylight.controller.hosttracker.IfIptoHost;
  90 import org.opendaylight.controller.hosttracker.hostAware.HostNodeConnector;
  91 import org.opendaylight.controller.switchmanager.ISwitchManager;
  92 import org.opendaylight.affinity.l2agent.L2Agent;
  93 import org.slf4j.Logger;
  94 import org.slf4j.LoggerFactory;
  95
  96 /**
  97  * The class caches latest network nodes statistics as notified by reader
  98  * services and provides API to retrieve them.
  99  */
 100 public class AffinityManagerImpl implements IAffinityManager, IConfigurationContainerAware, IObjectReader, ICacheUpdateAware<Long, String> {
 101     private static final Logger log = LoggerFactory.getLogger(AffinityManagerImpl.class);
 102
 103     private static String ROOT = GlobalConstants.STARTUPHOME.toString();
 104     private static final String SAVE = "Save";
 105     private String affinityLinkFileName = null;
 106     private String affinityGroupFileName = null;
 107     private IFlowProgrammerService fps = null;
 108     private ISwitchManager switchManager = null;
 109     private L2Agent l2agent = null;
 110
 111     private ConcurrentMap<String, AffinityGroup> affinityGroupList;
 112     private ConcurrentMap<String, AffinityLink> affinityLinkList;
 113     private ConcurrentMap<Long, String> configSaveEvent;
 114
 115     private IfIptoHost hostTracker;
 116
 117     private final Set<IAffinityManagerAware> affinityManagerAware = Collections
 118             .synchronizedSet(new HashSet<IAffinityManagerAware>());
 119
 120     private byte[] MAC;
 121     private static boolean hostRefresh = true;
 122     private int hostRetryCount = 5;
 123     private IClusterContainerServices clusterContainerService = null;
 124     private String containerName = null;
 125     private boolean isDefaultContainer = true;
 126     private static final int REPLACE_RETRY = 1;
 127
 128     public enum ReasonCode {
 129         SUCCESS("Success"), FAILURE("Failure"), INVALID_CONF(
 130                 "Invalid Configuration"), EXIST("Entry Already Exist"), CONFLICT(
 131                         "Configuration Conflict with Existing Entry");
 132
 133         private final String name;
 134
 135         private ReasonCode(String name) {
 136             this.name = name;
 137         }
 138
 139         @Override
 140         public String toString() {
 141             return name;
 142         }
 143     }
 144
 145     /* Only default container. */
 146     public String getContainerName() {
 147         return containerName;
 148     }
 149
 150     public void startUp() {
 151         // Initialize configuration file names
 152         affinityLinkFileName = ROOT + "affinityConfig_link" + this.getContainerName()
 153             + ".conf";
 154         affinityGroupFileName = ROOT + "affinityConfig_group" + this.getContainerName()
 155             + ".conf";
 156
 157         // Instantiate cluster synced variables
 158         allocateCaches();
 159         retrieveCaches();
 160
 161         /*
 162          * Read startup and build database if we have not already gotten the
 163          * configurations synced from another node
 164          */
 165         if (affinityGroupList.isEmpty() || affinityLinkList.isEmpty()) {
 166             loadAffinityConfiguration();
 167         }
 168     }
 169
 170     public void shutDown() {
 171     }
 172
 173     @SuppressWarnings("deprecation")
 174     private void allocateCaches() {
 175         if (this.clusterContainerService == null) {
 176             this.nonClusterObjectCreate();
 177             log.warn("un-initialized clusterContainerService, can't create cache");
 178             return;
 179         }
 180         try {
 181             clusterContainerService.createCache(
 182                     "affinity.affinityGroupList",
 183                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
 184             clusterContainerService.createCache(
 185                     "affinity.affinityLinkList",
 186                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
 187             clusterContainerService.createCache(
 188                     "affinity.configSaveEvent",
 189                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
 190         } catch (CacheConfigException cce) {
 191             log.error("\nCache configuration invalid - check cache mode");
 192         } catch (CacheExistException ce) {
 193             log.error("\nCache already exits - destroy and recreate if needed");
 194         }
 195     }
 196
 197     @SuppressWarnings({ "unchecked", "deprecation" })
 198     private void retrieveCaches() {
 199         if (this.clusterContainerService == null) {
 200             log.info("un-initialized clusterContainerService, can't retrieve cache");
 201             return;
 202         }
 203         affinityGroupList = (ConcurrentMap<String, AffinityGroup>) clusterContainerService
 204             .getCache("affinity.affinityGroupList");
 205         if (affinityGroupList == null) {
 206             log.error("\nFailed to get cache for affinityGroupList");
 207         }
 208         affinityLinkList = (ConcurrentMap<String, AffinityLink>) clusterContainerService
 209             .getCache("affinity.affinityLinkList");
 210         if (affinityLinkList == null) {
 211             log.error("\nFailed to get cache for affinityLinkList");
 212         }
 213
 214         configSaveEvent = (ConcurrentMap<Long, String>) clusterContainerService
 215             .getCache("affinity.configSaveEvent");
 216         if (configSaveEvent == null) {
 217             log.error("\nFailed to get cache for configSaveEvent");
 218         }
 219     }
 220
 221     private void nonClusterObjectCreate() {
 222         affinityLinkList = new ConcurrentHashMap<String, AffinityLink>();
 223         affinityGroupList = new ConcurrentHashMap<String, AffinityGroup>();
 224         configSaveEvent = new ConcurrentHashMap<Long, String>();
 225     }
 226
 227
 228     void setHostTracker(IfIptoHost h) {
 229         this.hostTracker = h;
 230     }
 231
 232     void unsetHostTracker(IfIptoHost h) {
 233         if (this.hostTracker.equals(h)) {
 234             this.hostTracker = null;
 235         }
 236     }
 237     public void setFlowProgrammerService(IFlowProgrammerService s)
 238     {
 239         this.fps = s;
 240     }
 241
 242     public void unsetFlowProgrammerService(IFlowProgrammerService s) {
 243         if (this.fps == s) {
 244             this.fps = null;
 245         }
 246     }
 247 <<<<<<< HEAD
 248     public void setL2Agent(L2Agent s)
 249 =======
 250     public void setL2Agent(TutorialL2Forwarding s)
 251 >>>>>>> Add flow pushing rules.
 252     {
 253         this.l2agent = s;
 254     }
 255
 256 <<<<<<< HEAD
 257     public void unsetL2Agent(L2Agent s) {
 258 =======
 259     public void unsetL2Agent(TutorialL2Forwarding s) {
 260 >>>>>>> Add flow pushing rules.
 261         if (this.l2agent == s) {
 262             this.l2agent = null;
 263         }
 264     }
 265
 266     /*
 267     public void setForwardingRulesManager(
 268             IForwardingRulesManager forwardingRulesManager) {
 269         this.ruleManager = forwardingRulesManager;
 270     }
 271
 272     public void unsetForwardingRulesManager(
 273             IForwardingRulesManager forwardingRulesManager) {
 274         if (this.ruleManager == forwardingRulesManager) {
 275             this.ruleManager = null;
 276         }
 277     }
 278     */
 279
 280     public Status addAffinityLink(AffinityLink al) {
 281         boolean putNewLink = false;
 282
 283         if (affinityLinkList.containsKey(al.getName())) {
 284             return new Status(StatusCode.CONFLICT,
 285                               "AffinityLink with the specified name already configured.");
 286         }
 287
 288
 289         AffinityLink alCurr = affinityLinkList.get(al.getName());
 290         if (alCurr == null) {
 291             if (affinityLinkList.putIfAbsent(al.getName(), al) == null) {
 292                 putNewLink = true;
 293             }
 294         } else {
 295             putNewLink = affinityLinkList.replace(al.getName(), alCurr, al);
 296         }
 297
 298         if (!putNewLink) {
 299             String msg = "Cluster conflict: Conflict while adding the subnet " + al.getName();
 300             return new Status(StatusCode.CONFLICT, msg);
 301         }
 302
 303         return new Status(StatusCode.SUCCESS);
 304     }
 305
 306
 307     /**
 308      * Fetch all node connectors. Each switch port will receive a flow rule. Do not stop on error.
 309      */
 310     public Status pushFlowRule(Flow flow) {
 311         /* Get all node connectors. */
 312         Set<Node> nodes = switchManager.getNodes();
 313         Status success = new Status(StatusCode.SUCCESS);
 314         Status notfound = new Status(StatusCode.NOTFOUND);
 315
 316         if (nodes == null) {
 317             log.debug("No nodes in network.");
 318             return success;
 319         }
 320         for (Node node: nodes) {
 321             Set<NodeConnector> ncs = switchManager.getNodeConnectors(node);
 322             if (ncs == null) {
 323                 continue;
 324             }
 325             Status status = fps.addFlow(node, flow);
 326             if (!status.isSuccess()) {
 327                 log.debug("Error during addFlow: {} on {}. The failure is: {}",
 328                           flow, node, status.getDescription());
 329             }
 330         }
 331         return success;
 332     }
 333
 334     /**
 335      * add flow rules for each node connector.
 336      */
 337     public Status addFlowRulesForRedirect(AffinityLink al) throws Exception {
 338         Match match = new Match();
 339         List<Action> actions = new ArrayList<Action>();
 340
 341         InetAddress address1, address2;
 342         InetAddress mask;
 343         mask = InetAddress.getByName("255.255.255.255");
 344
 345         Flow f = new Flow(match, actions);
 346
 347         List<Entry<Host,Host>> hostPairList= getAllFlowsByHost(al);
 348         for (Entry<Host,Host> hostPair : hostPairList) {
 349             /* Create a match for each host pair in the affinity link. */
 350
 351             Host host1 = hostPair.getKey();
 352             Host host2 = hostPair.getValue();
 353             address1 = host1.getNetworkAddress();
 354             address2 = host2.getNetworkAddress();
 355
 356             match.setField(MatchType.NW_SRC, address1, mask);
 357             match.setField(MatchType.NW_DST, address2, mask);
 358
 359
 360             /* For each end point, discover the mac address of the
 361              * host. Then lookup the L2 table to find the port to send
 362              * this flow along. Program the flow. */
 363
 364             byte [] mac = ((HostNodeConnector) host1).getDataLayerAddressBytes();
 365             /* Tbd: use hosttracker for this. */
 366             //            NodeConnector dst_connector = l2agent.lookupMacAddress(mac);
 367             //            actions.add(new Output(dst_connector));
 368         }
 369         return new Status(StatusCode.SUCCESS);
 370     }
 371
 372     public Status removeAffinityLink(String name) {
 373         affinityLinkList.remove(name);
 374         return new Status(StatusCode.SUCCESS);
 375     }
 376
 377     public Status removeAffinityLink(AffinityLink al) {
 378         AffinityLink alCurr = affinityLinkList.get(al.getName());
 379         if (alCurr != null) {
 380             affinityLinkList.remove(alCurr);
 381             return new Status(StatusCode.SUCCESS);
 382         } else {
 383             String msg = "Affinity Link with specified name does not exist." + al.getName();
 384             return new Status(StatusCode.INTERNALERROR, msg);
 385         }
 386     }
 387
 388     @Override
 389     public AffinityLink getAffinityLink(String linkName) {
 390         return affinityLinkList.get(linkName);
 391     }
 392
 393     @Override
 394     public List<AffinityLink> getAllAffinityLinks() {
 395         return new ArrayList<AffinityLink>(affinityLinkList.values());
 396     }
 397
 398     @Override
 399     public Status addAffinityGroup(AffinityGroup ag) {
 400         boolean putNewGroup = false;
 401         String name = ag.getName();
 402         if (affinityGroupList.containsKey(name)) {
 403             return new Status(StatusCode.CONFLICT,
 404                               "AffinityGroup with the specified name already configured.");
 405         }
 406         AffinityGroup agCurr = affinityGroupList.get(name);
 407         if (agCurr == null) {
 408             if (affinityGroupList.putIfAbsent(name, ag) == null) {
 409                 putNewGroup = true;
 410             }
 411         } else {
 412             putNewGroup = affinityGroupList.replace(name, agCurr, ag);
 413         }
 414
 415         if (!putNewGroup) {
 416             String msg = "Cluster conflict: Conflict while adding the subnet " + name;
 417             return new Status(StatusCode.CONFLICT, msg);
 418         }
 419
 420         return new Status(StatusCode.SUCCESS);
 421     }
 422
 423     /* Check for errors. */
 424     @Override
 425     public Status removeAffinityGroup(String name) {
 426         affinityGroupList.remove(name);
 427         return new Status(StatusCode.SUCCESS);
 428     }
 429
 430     @Override
 431     public AffinityGroup getAffinityGroup(String groupName) {
 432         return affinityGroupList.get(groupName);
 433     }
 434
 435     @Override
 436     public List<AffinityGroup> getAllAffinityGroups() {
 437         return new ArrayList<AffinityGroup>(affinityGroupList.values());
 438     }
 439
 440     /* Find where this is used. */
 441     @Override
 442     public Object readObject(ObjectInputStream ois)
 443             throws FileNotFoundException, IOException, ClassNotFoundException {
 444         // Perform the class deserialization locally, from inside the package
 445         // where the class is defined
 446         return ois.readObject();
 447     }
 448
 449     @SuppressWarnings("unchecked")
 450     private void loadAffinityConfiguration() {
 451         ObjectReader objReader = new ObjectReader();
 452         ConcurrentMap<String, AffinityGroup> groupList = (ConcurrentMap<String, AffinityGroup>) objReader.read(this, affinityGroupFileName);
 453         ConcurrentMap<String, AffinityLink> linkList = (ConcurrentMap<String, AffinityLink>) objReader.read(this, affinityLinkFileName);
 454
 455         /* group list */
 456         if (groupList != null) {
 457             for (AffinityGroup ag : groupList.values()) {
 458                 addAffinityGroup(ag);
 459             }
 460         }
 461
 462         /* link list */
 463         if (linkList != null) {
 464             for (AffinityLink al : linkList.values()) {
 465                 addAffinityLink(al);
 466             }
 467         }
 468     }
 469
 470     @Override
 471     public ArrayList<AffinityIdentifier> getAllElementsByAffinityIdentifier(AffinityGroup ag) {
 472         return ag.getAllElements();
 473     }
 474
 475     @Override
 476     public List<Host> getAllElementsByHost(AffinityGroup ag) {
 477         List<Host> hostList= new ArrayList<Host>();
 478
 479         for (AffinityIdentifier h : ag.getAllElements()) {
 480             /* TBD: Do not assume this to be an InetAddress. */
 481             h.print();
 482             if (hostTracker != null) {
 483                 Host host1 = hostTracker.hostFind((InetAddress) h.get());
 484                 hostList.add(host1);
 485             }
 486         }
 487         return hostList;
 488     }
 489
 490     @Override
 491     public List<Entry<Host, Host>> getAllFlowsByHost(AffinityLink al) {
 492         List<Entry<Host,Host>> hostPairList= new ArrayList<Entry<Host, Host>>();
 493
 494         AffinityGroup fromGroup = al.getFromGroup();
 495         AffinityGroup toGroup = al.getToGroup();
 496
 497         for (AffinityIdentifier h1 : fromGroup.getAllElements()) {
 498             for (AffinityIdentifier h2 : toGroup.getAllElements()) {
 499                 if (hostTracker != null) {
 500                     Host host1 = hostTracker.hostFind((InetAddress) h1.get());
 501                     Host host2 = hostTracker.hostFind((InetAddress) h2.get());
 502                     Entry<Host, Host> hp1=new AbstractMap.SimpleEntry<Host, Host>(host1, host2);
 503                     hostPairList.add(hp1);
 504                 }
 505             }
 506         }
 507         return hostPairList;
 508     }
 509
 510     @Override
 511     public List<Entry<AffinityIdentifier, AffinityIdentifier>> getAllFlowsByAffinityIdentifier(AffinityLink al) {
 512         List<Entry<AffinityIdentifier, AffinityIdentifier>> hostPairList= new ArrayList<Entry<AffinityIdentifier, AffinityIdentifier>>();
 513
 514         AffinityGroup fromGroup = al.getFromGroup();
 515         AffinityGroup toGroup = al.getToGroup();
 516
 517         for (AffinityIdentifier h1 : fromGroup.getAllElements()) {
 518             for (AffinityIdentifier h2 : toGroup.getAllElements()) {
 519                 Entry<AffinityIdentifier, AffinityIdentifier> hp1=new AbstractMap.SimpleEntry<AffinityIdentifier, AffinityIdentifier>(h1, h2);
 520                 hostPairList.add(hp1);
 521             }
 522         }
 523         return hostPairList;
 524     }
 525
 526     @Override
 527     public Status saveConfiguration() {
 528         return saveAffinityConfig();
 529     }
 530
 531     @Override
 532     public Status saveAffinityConfig() {
 533         // Publish the save config event to the cluster nodes
 534         configSaveEvent.put(new Date().getTime(), SAVE);
 535         return saveAffinityConfigInternal();
 536     }
 537
 538     public Status saveAffinityConfigInternal() {
 539         Status retS = null, retP = null;
 540         ObjectWriter objWriter = new ObjectWriter();
 541
 542         retS = objWriter.write(new ConcurrentHashMap<String, AffinityLink>(
 543                 affinityLinkList), affinityLinkFileName);
 544
 545         retP = objWriter.write(new ConcurrentHashMap<String, AffinityGroup>(
 546                 affinityGroupList), affinityGroupFileName);
 547
 548         if (retS.isSuccess() && retP.isSuccess()) {
 549             return new Status(StatusCode.SUCCESS, "Configuration saved.");
 550         } else {
 551             return new Status(StatusCode.INTERNALERROR, "Save failed");
 552         }
 553     }
 554
 555     @Override
 556     public void entryCreated(Long key, String cacheName, boolean local) {
 557     }
 558
 559     @Override
 560     public void entryUpdated(Long key, String new_value, String cacheName,
 561             boolean originLocal) {
 562         saveAffinityConfigInternal();
 563     }
 564
 565     @Override
 566     public void entryDeleted(Long key, String cacheName, boolean originLocal) {
 567     }
 568
 569     /**
 570      * Function called by the dependency manager when all the required
 571      * dependencies are satisfied
 572      *
 573      */
 574     void init() {
 575         log.debug("INIT called!");
 576         containerName = GlobalConstants.DEFAULT.toString();
 577         startUp();
 578     }
 579
 580     /**
 581      * Function called by the dependency manager when at least one
 582      * dependency become unsatisfied or when the component is shutting
 583      * down because for example bundle is being stopped.
 584      *
 585      */
 586     void destroy() {
 587         log.debug("DESTROY called!");
 588     }
 589
 590     /**
 591      * Function called by dependency manager after "init ()" is called
 592      * and after the services provided by the class are registered in
 593      * the service registry
 594      *
 595      */
 596     void start() {
 597         log.debug("START called!");
 598     }
 599
 600     /**
 601      * Function called after registering the service in OSGi service registry.
 602      */
 603     void started(){
 604         // Retrieve current statistics so we don't have to wait for next refresh
 605         IAffinityManager affinityManager = (IAffinityManager) ServiceHelper.getInstance(
 606                 IAffinityManager.class, this.getContainerName(), this);
 607         if (affinityManager != null) {
 608             log.debug("STARTED method called!");
 609         }
 610     }
 611
 612     /**
 613      * Function called by the dependency manager before the services
 614      * exported by the component are unregistered, this will be
 615      * followed by a "destroy ()" calls
 616      *
 617      */
 618     void stop() {
 619         log.debug("STOP called!");
 620     }
 621
 622     void setClusterContainerService(IClusterContainerServices s) {
 623         log.debug("Cluster Service set for affinity mgr");
 624         this.clusterContainerService = s;
 625     }
 626
 627     void unsetClusterContainerService(IClusterContainerServices s) {
 628         if (this.clusterContainerService == s) {
 629             log.debug("Cluster Service removed for affinity mgr!");
 630             this.clusterContainerService = null;
 631         }
 632     }
 633 }