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