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