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