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