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