opendaylight/md-sal/forwardingrules-manager/src/main/java/org/opendaylight/controller/forwardingrulesmanager/consumer/impl/GroupConsumerImpl.java

   1 package org.opendaylight.controller.forwardingrulesmanager.consumer.impl;
   2
   3 import java.util.ArrayList;
   4 import java.util.Collection;
   5 import java.util.EnumSet;
   6 import java.util.HashMap;
   7 import java.util.HashSet;
   8 import java.util.Iterator;
   9 import java.util.List;
  10 import java.util.Map;
  11 import java.util.Map.Entry;
  12 import java.util.Set;
  13 import java.util.concurrent.ConcurrentHashMap;
  14 import java.util.concurrent.ConcurrentMap;
  15
  16 import org.opendaylight.controller.clustering.services.CacheConfigException;
  17 import org.opendaylight.controller.clustering.services.CacheExistException;
  18 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
  19 import org.opendaylight.controller.clustering.services.IClusterServices;
  20 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler;
  21 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction;
  22 import org.opendaylight.controller.md.sal.common.api.data.DataModification;
  23 import org.opendaylight.controller.sal.common.util.Rpcs;
  24 import org.opendaylight.controller.sal.core.IContainer;
  25 import org.opendaylight.controller.sal.core.Node;
  26 import org.opendaylight.controller.sal.utils.GlobalConstants;
  27 import org.opendaylight.controller.sal.utils.Status;
  28 import org.opendaylight.controller.sal.utils.StatusCode;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.config.rev131024.Groups;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.config.rev131024.groups.Group;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.config.rev131024.groups.GroupKey;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.AddGroupInputBuilder;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.GroupAdded;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.GroupRemoved;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.GroupUpdated;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.SalGroupListener;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.SalGroupService;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.UpdateGroupInputBuilder;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.group.update.UpdatedGroupBuilder;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupTypes.GroupType;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.Buckets;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.buckets.Bucket;
  43 import org.opendaylight.yangtools.concepts.Registration;
  44 import org.opendaylight.yangtools.yang.binding.DataObject;
  45 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  46 import org.opendaylight.yangtools.yang.binding.NotificationListener;
  47 import org.opendaylight.yangtools.yang.common.RpcResult;
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 @SuppressWarnings("unused")
  52 public class GroupConsumerImpl implements IForwardingRulesManager {
  53
  54     protected static final Logger logger = LoggerFactory.getLogger(GroupConsumerImpl.class);
  55     private final GroupEventListener groupEventListener = new GroupEventListener();
  56     private Registration<NotificationListener> groupListener;
  57     private SalGroupService groupService;
  58     private GroupDataCommitHandler commitHandler;
  59
  60     private ConcurrentMap<GroupKey, Group> originalSwGroupView;
  61     private ConcurrentMap<GroupKey, Group> installedSwGroupView;
  62
  63     private ConcurrentMap<Node, List<Group>> nodeGroups;
  64     private ConcurrentMap<GroupKey, Group> inactiveGroups;
  65
  66     private IClusterContainerServices clusterGroupContainerService = null;
  67     private IContainer container;
  68
  69     public GroupConsumerImpl() {
  70
  71         InstanceIdentifier<? extends DataObject> path = InstanceIdentifier.builder().node(Groups.class)
  72                 .node(Group.class).toInstance();
  73         groupService = FRMConsumerImpl.getProviderSession().getRpcService(SalGroupService.class);
  74
  75         clusterGroupContainerService = FRMConsumerImpl.getClusterContainerService();
  76         container = FRMConsumerImpl.getContainer();
  77
  78         if (!(cacheStartup())) {
  79             logger.error("Unanle to allocate/retrieve group cache");
  80             System.out.println("Unable to allocate/retrieve group cache");
  81         }
  82
  83         if (null == groupService) {
  84             logger.error("Consumer SAL Group Service is down or NULL. FRM may not function as intended");
  85             System.out.println("Consumer SAL Group Service is down or NULL.");
  86             return;
  87         }
  88
  89         // For switch events
  90         groupListener = FRMConsumerImpl.getNotificationService().registerNotificationListener(groupEventListener);
  91
  92         if (null == groupListener) {
  93             logger.error("Listener to listen on group data modifcation events");
  94             System.out.println("Listener to listen on group data modifcation events.");
  95             return;
  96         }
  97
  98         commitHandler = new GroupDataCommitHandler();
  99         FRMConsumerImpl.getDataProviderService().registerCommitHandler(path, commitHandler);
 100     }
 101
 102     private boolean allocateGroupCaches() {
 103         if (this.clusterGroupContainerService == null) {
 104             logger.warn("Group: Un-initialized clusterGroupContainerService, can't create cache");
 105             return false;
 106         }
 107
 108         try {
 109             clusterGroupContainerService.createCache("frm.originalSwGroupView",
 110                     EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 111
 112             clusterGroupContainerService.createCache("frm.installedSwGroupView",
 113                     EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 114
 115             clusterGroupContainerService.createCache("frm.inactiveGroups",
 116                     EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 117
 118             clusterGroupContainerService.createCache("frm.nodeGroups",
 119                     EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 120
 121             // TODO for cluster mode
 122             /*
 123              * clusterGroupContainerService.createCache(WORK_STATUS_CACHE,
 124              * EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL,
 125              * IClusterServices.cacheMode.ASYNC));
 126              *
 127              * clusterGroupContainerService.createCache(WORK_ORDER_CACHE,
 128              * EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL,
 129              * IClusterServices.cacheMode.ASYNC));
 130              */
 131
 132         } catch (CacheConfigException cce) {
 133             logger.error("Group CacheConfigException");
 134             return false;
 135
 136         } catch (CacheExistException cce) {
 137             logger.error(" Group CacheExistException");
 138         }
 139
 140         return true;
 141     }
 142
 143     private void nonClusterGroupObjectCreate() {
 144         originalSwGroupView = new ConcurrentHashMap<GroupKey, Group>();
 145         installedSwGroupView = new ConcurrentHashMap<GroupKey, Group>();
 146         nodeGroups = new ConcurrentHashMap<Node, List<Group>>();
 147         inactiveGroups = new ConcurrentHashMap<GroupKey, Group>();
 148     }
 149
 150     @SuppressWarnings({ "unchecked" })
 151     private boolean retrieveGroupCaches() {
 152         ConcurrentMap<?, ?> map;
 153
 154         if (this.clusterGroupContainerService == null) {
 155             logger.warn("Group: un-initialized clusterGroupContainerService, can't retrieve cache");
 156             nonClusterGroupObjectCreate();
 157             return false;
 158         }
 159
 160         map = clusterGroupContainerService.getCache("frm.originalSwGroupView");
 161         if (map != null) {
 162             originalSwGroupView = (ConcurrentMap<GroupKey, Group>) map;
 163         } else {
 164             logger.error("Retrieval of cache(originalSwGroupView) failed");
 165             return false;
 166         }
 167
 168         map = clusterGroupContainerService.getCache("frm.installedSwGroupView");
 169         if (map != null) {
 170             installedSwGroupView = (ConcurrentMap<GroupKey, Group>) map;
 171         } else {
 172             logger.error("Retrieval of cache(installedSwGroupView) failed");
 173             return false;
 174         }
 175
 176         map = clusterGroupContainerService.getCache("frm.inactiveGroups");
 177         if (map != null) {
 178             inactiveGroups = (ConcurrentMap<GroupKey, Group>) map;
 179         } else {
 180             logger.error("Retrieval of cache(inactiveGroups) failed");
 181             return false;
 182         }
 183
 184         map = clusterGroupContainerService.getCache("frm.nodeGroups");
 185         if (map != null) {
 186             nodeGroups = (ConcurrentMap<Node, List<Group>>) map;
 187         } else {
 188             logger.error("Retrieval of cache(nodeGroup) failed");
 189             return false;
 190         }
 191
 192         return true;
 193     }
 194
 195     private boolean cacheStartup() {
 196         if (allocateGroupCaches()) {
 197             if (retrieveGroupCaches()) {
 198                 return true;
 199             }
 200         }
 201
 202         return false;
 203     }
 204
 205     public Status validateGroup(Group group, FRMUtil.operation operation) {
 206         String containerName;
 207         String groupName;
 208         Iterator<Bucket> bucketIterator;
 209         boolean returnResult;
 210         Buckets groupBuckets;
 211
 212         if (null != group) {
 213             containerName = group.getContainerName();
 214
 215             if (null == containerName) {
 216                 containerName = GlobalConstants.DEFAULT.toString();
 217             } else if (!FRMUtil.isNameValid(containerName)) {
 218                 logger.error("Container Name is invalid %s" + containerName);
 219                 return new Status(StatusCode.BADREQUEST, "Container Name is invalid");
 220             }
 221
 222             groupName = group.getGroupName();
 223             if (!FRMUtil.isNameValid(groupName)) {
 224                 logger.error("Group Name is invalid %s" + groupName);
 225                 return new Status(StatusCode.BADREQUEST, "Group Name is invalid");
 226             }
 227
 228             returnResult = doesGroupEntryExists(group.getKey(), groupName, containerName);
 229
 230             if (FRMUtil.operation.ADD == operation && returnResult) {
 231                 logger.error("Record with same Group Name exists");
 232                 return new Status(StatusCode.BADREQUEST, "Group record exists");
 233             } else if (!returnResult) {
 234                 logger.error("Group record does not exist");
 235                 return new Status(StatusCode.BADREQUEST, "Group record does not exist");
 236             }
 237
 238             if (!(group.getGroupType().getIntValue() >= GroupType.GroupAll.getIntValue() && group.getGroupType()
 239                     .getIntValue() <= GroupType.GroupFf.getIntValue())) {
 240                 logger.error("Invalid Group type %d" + group.getGroupType().getIntValue());
 241                 return new Status(StatusCode.BADREQUEST, "Invalid Group type");
 242             }
 243
 244             groupBuckets = group.getBuckets();
 245
 246             if (null != groupBuckets && null != groupBuckets.getBucket()) {
 247                 bucketIterator = groupBuckets.getBucket().iterator();
 248
 249                 while (bucketIterator.hasNext()) {
 250                     if (!(FRMUtil.validateActions(bucketIterator.next().getAction()))) {
 251                         logger.error("Error in action bucket");
 252                         return new Status(StatusCode.BADREQUEST, "Invalid Group bucket contents");
 253                     }
 254                 }
 255             }
 256         }
 257
 258         return new Status(StatusCode.SUCCESS);
 259
 260     }
 261
 262     private boolean doesGroupEntryExists(GroupKey key, String groupName, String containerName) {
 263         if (!originalSwGroupView.containsKey(key)) {
 264             return false;
 265         }
 266
 267         for (ConcurrentMap.Entry<GroupKey, Group> entry : originalSwGroupView.entrySet()) {
 268             if (entry.getValue().getGroupName().equals(groupName)) {
 269                 if (entry.getValue().getContainerName().equals(containerName)) {
 270                     return true;
 271                 }
 272             }
 273         }
 274         return false;
 275     }
 276
 277     /**
 278      * Update Group entries to the southbound plugin/inventory and our internal
 279      * database
 280      *
 281      * @param path
 282      * @param dataObject
 283      */
 284     private Status updateGroup(InstanceIdentifier<?> path, Group groupUpdateDataObject) {
 285         GroupKey groupKey = groupUpdateDataObject.getKey();
 286         UpdatedGroupBuilder updateGroupBuilder = null;
 287
 288         Status groupOperationStatus = validateGroup(groupUpdateDataObject, FRMUtil.operation.UPDATE);
 289
 290         if (!groupOperationStatus.isSuccess()) {
 291             logger.error("Group data object validation failed %s" + groupUpdateDataObject.getGroupName());
 292             return groupOperationStatus;
 293         }
 294
 295         if (originalSwGroupView.containsKey(groupKey)) {
 296             originalSwGroupView.remove(groupKey);
 297             originalSwGroupView.put(groupKey, groupUpdateDataObject);
 298         }
 299
 300         if (groupUpdateDataObject.isInstall()) {
 301             UpdateGroupInputBuilder groupData = new UpdateGroupInputBuilder();
 302             updateGroupBuilder = new UpdatedGroupBuilder();
 303             updateGroupBuilder.fieldsFrom(groupUpdateDataObject);
 304             groupData.setUpdatedGroup(updateGroupBuilder.build());
 305             // TODO how to get original group and modified group.
 306
 307             if (installedSwGroupView.containsKey(groupKey)) {
 308                 installedSwGroupView.remove(groupKey);
 309                 installedSwGroupView.put(groupKey, groupUpdateDataObject);
 310             }
 311
 312             groupService.updateGroup(groupData.build());
 313         }
 314
 315         return groupOperationStatus;
 316     }
 317
 318     /**
 319      * Adds Group to the southbound plugin and our internal database
 320      *
 321      * @param path
 322      * @param dataObject
 323      */
 324     private Status addGroup(InstanceIdentifier<?> path, Group groupAddDataObject) {
 325         GroupKey groupKey = groupAddDataObject.getKey();
 326         Status groupOperationStatus = validateGroup(groupAddDataObject, FRMUtil.operation.ADD);
 327
 328         if (!groupOperationStatus.isSuccess()) {
 329             logger.error("Group data object validation failed %s" + groupAddDataObject.getGroupName());
 330             return groupOperationStatus;
 331         }
 332
 333         originalSwGroupView.put(groupKey, groupAddDataObject);
 334
 335         if (groupAddDataObject.isInstall()) {
 336             AddGroupInputBuilder groupData = new AddGroupInputBuilder();
 337             groupData.setBuckets(groupAddDataObject.getBuckets());
 338             groupData.setContainerName(groupAddDataObject.getContainerName());
 339             groupData.setGroupId(groupAddDataObject.getGroupId());
 340             groupData.setGroupType(groupAddDataObject.getGroupType());
 341             groupData.setNode(groupAddDataObject.getNode());
 342             installedSwGroupView.put(groupKey, groupAddDataObject);
 343             groupService.addGroup(groupData.build());
 344         }
 345
 346         return groupOperationStatus;
 347     }
 348
 349     private RpcResult<Void> commitToPlugin(internalTransaction transaction) {
 350         for (Entry<InstanceIdentifier<?>, Group> entry : transaction.additions.entrySet()) {
 351
 352             if (!addGroup(entry.getKey(), entry.getValue()).isSuccess()) {
 353                 transaction.additions.remove(entry.getKey());
 354                 return Rpcs.getRpcResult(false, null, null);
 355             }
 356         }
 357
 358         for (Entry<InstanceIdentifier<?>, Group> entry : transaction.updates.entrySet()) {
 359
 360             if (!updateGroup(entry.getKey(), entry.getValue()).isSuccess()) {
 361                 transaction.updates.remove(entry.getKey());
 362                 return Rpcs.getRpcResult(false, null, null);
 363             }
 364         }
 365
 366         for (InstanceIdentifier<?> removal : transaction.removals) {
 367             // removeFlow(removal);
 368         }
 369
 370         return Rpcs.getRpcResult(true, null, null);
 371     }
 372
 373     private final class GroupDataCommitHandler implements DataCommitHandler<InstanceIdentifier<?>, DataObject> {
 374
 375         @SuppressWarnings("unchecked")
 376         @Override
 377         public DataCommitTransaction<InstanceIdentifier<?>, DataObject> requestCommit(
 378                 DataModification<InstanceIdentifier<?>, DataObject> modification) {
 379             // We should verify transaction
 380             System.out.println("Coming in GroupDatacommitHandler");
 381             internalTransaction transaction = new internalTransaction(modification);
 382             transaction.prepareUpdate();
 383             return transaction;
 384         }
 385     }
 386
 387     private final class internalTransaction implements DataCommitTransaction<InstanceIdentifier<?>, DataObject> {
 388
 389         private final DataModification<InstanceIdentifier<?>, DataObject> modification;
 390
 391         @Override
 392         public DataModification<InstanceIdentifier<?>, DataObject> getModification() {
 393             return modification;
 394         }
 395
 396         public internalTransaction(DataModification<InstanceIdentifier<?>, DataObject> modification) {
 397             this.modification = modification;
 398         }
 399
 400         Map<InstanceIdentifier<?>, Group> additions = new HashMap<>();
 401         Map<InstanceIdentifier<?>, Group> updates = new HashMap<>();
 402         Set<InstanceIdentifier<?>> removals = new HashSet<>();
 403
 404         /**
 405          * We create a plan which flows will be added, which will be updated and
 406          * which will be removed based on our internal state.
 407          *
 408          */
 409         void prepareUpdate() {
 410
 411             Set<Entry<InstanceIdentifier<?>, DataObject>> puts = modification.getUpdatedConfigurationData().entrySet();
 412             for (Entry<InstanceIdentifier<?>, DataObject> entry : puts) {
 413                 if (entry.getValue() instanceof Group) {
 414                     Group group = (Group) entry.getValue();
 415                     preparePutEntry(entry.getKey(), group);
 416                 }
 417
 418             }
 419
 420             removals = modification.getRemovedConfigurationData();
 421         }
 422
 423         private void preparePutEntry(InstanceIdentifier<?> key, Group group) {
 424
 425             Group original = originalSwGroupView.get(key);
 426             if (original != null) {
 427                 // It is update for us
 428
 429                 updates.put(key, group);
 430             } else {
 431                 // It is addition for us
 432
 433                 additions.put(key, group);
 434             }
 435         }
 436
 437         /**
 438          * We are OK to go with execution of plan
 439          *
 440          */
 441         @Override
 442         public RpcResult<Void> finish() throws IllegalStateException {
 443
 444             RpcResult<Void> rpcStatus = commitToPlugin(this);
 445             // We return true if internal transaction is successful.
 446             // return Rpcs.getRpcResult(true, null, Collections.emptySet());
 447             return rpcStatus;
 448         }
 449
 450         /**
 451          *
 452          * We should rollback our preparation
 453          *
 454          */
 455         @Override
 456         public RpcResult<Void> rollback() throws IllegalStateException {
 457             // NOOP - we did not modified any internal state during
 458             // requestCommit phase
 459             // return Rpcs.getRpcResult(true, null, Collections.emptySet());
 460             return Rpcs.getRpcResult(true, null, null);
 461
 462         }
 463
 464     }
 465
 466     final class GroupEventListener implements SalGroupListener {
 467
 468         List<GroupAdded> addedGroups = new ArrayList<>();
 469         List<GroupRemoved> removedGroups = new ArrayList<>();
 470         List<GroupUpdated> updatedGroups = new ArrayList<>();
 471
 472         @Override
 473         public void onGroupAdded(GroupAdded notification) {
 474             System.out.println("added Group..........................");
 475             addedGroups.add(notification);
 476         }
 477
 478         @Override
 479         public void onGroupRemoved(GroupRemoved notification) {
 480             // TODO Auto-generated method stub
 481
 482         }
 483
 484         @Override
 485         public void onGroupUpdated(GroupUpdated notification) {
 486             // TODO Auto-generated method stub
 487
 488         }
 489     }
 490
 491     @Override
 492     public List<DataObject> get() {
 493
 494         List<DataObject> orderedList = new ArrayList<DataObject>();
 495         Collection<Group> groupList = originalSwGroupView.values();
 496         for (Iterator<Group> iterator = groupList.iterator(); iterator.hasNext();) {
 497             orderedList.add(iterator.next());
 498         }
 499         return orderedList;
 500     }
 501
 502     @Override
 503     public DataObject getWithName(String name, Node n) {
 504
 505         if (this instanceof GroupConsumerImpl) {
 506             Collection<Group> groupList = originalSwGroupView.values();
 507             for (Iterator<Group> iterator = groupList.iterator(); iterator.hasNext();) {
 508                 Group group = iterator.next();
 509                 if (group.getNode().equals(n) && group.getGroupName().equals(name)) {
 510
 511                     return group;
 512                 }
 513             }
 514         }
 515         return null;
 516     }
 517 }