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