applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/ForwardingRulesManagerImpl.java

   1 /**
   2  * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.openflowplugin.applications.frm.impl;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import java.util.Objects;
  15 import java.util.Optional;
  16 import java.util.concurrent.ExecutionException;
  17 import java.util.concurrent.atomic.AtomicLong;
  18 import javax.annotation.Nonnull;
  19 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  20 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  21 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  22 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  23 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  24 import org.opendaylight.openflowplugin.api.openflow.configuration.ConfigurationService;
  25 import org.opendaylight.openflowplugin.api.openflow.mastership.MastershipChangeServiceManager;
  26 import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation;
  27 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesCommiter;
  28 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
  29 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesProperty;
  30 import org.opendaylight.openflowplugin.applications.frm.NodeConfigurator;
  31 import org.opendaylight.openflowplugin.applications.frm.nodeconfigurator.NodeConfiguratorImpl;
  32 import org.opendaylight.openflowplugin.applications.frm.recovery.OpenflowServiceRecoveryHandler;
  33 import org.opendaylight.openflowplugin.applications.reconciliation.NotificationRegistration;
  34 import org.opendaylight.openflowplugin.applications.reconciliation.ReconciliationManager;
  35 import org.opendaylight.serviceutils.srm.RecoverableListener;
  36 import org.opendaylight.serviceutils.srm.ServiceRecoveryRegistry;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.SalGroupService;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.service.rev130918.SalMeterService;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.SalBundleService;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.app.arbitrator.reconcile.service.rev180227.ArbitratorReconcileService;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.app.forwardingrules.manager.config.rev160511.ForwardingRulesManagerConfig;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.app.frm.reconciliation.service.rev180227.FrmReconciliationService;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.service.rev131026.SalTableService;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeatures;
  52 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55
  56 /**
  57  * forwardingrules-manager org.opendaylight.openflowplugin.applications.frm.impl
  58  *
  59  * <p>
  60  * Manager and middle point for whole module. It contains ActiveNodeHolder and
  61  * provide all RPC services.
  62  *
  63  */
  64 public class ForwardingRulesManagerImpl implements ForwardingRulesManager {
  65     private static final Logger LOG = LoggerFactory.getLogger(ForwardingRulesManagerImpl.class);
  66
  67     static final int STARTUP_LOOP_TICK = 500;
  68     static final int STARTUP_LOOP_MAX_RETRIES = 8;
  69     private static final int FRM_RECONCILIATION_PRIORITY = Integer.getInteger("frm.reconciliation.priority", 1);
  70     private static final String SERVICE_NAME = "FRM";
  71
  72     private final AtomicLong txNum = new AtomicLong();
  73     private final DataBroker dataService;
  74     private final SalFlowService salFlowService;
  75     private final SalGroupService salGroupService;
  76     private final SalMeterService salMeterService;
  77     private final SalTableService salTableService;
  78     private final ClusterSingletonServiceProvider clusterSingletonServiceProvider;
  79     private final SalBundleService salBundleService;
  80     private final AutoCloseable configurationServiceRegistration;
  81     private final MastershipChangeServiceManager mastershipChangeServiceManager;
  82     private final RpcProviderRegistry rpcRegistry;
  83     private ForwardingRulesCommiter<Flow> flowListener;
  84     private ForwardingRulesCommiter<Group> groupListener;
  85     private ForwardingRulesCommiter<Meter> meterListener;
  86     private ForwardingRulesCommiter<TableFeatures> tableListener;
  87     private FlowNodeReconciliation nodeListener;
  88     private NotificationRegistration reconciliationNotificationRegistration;
  89     private FlowNodeConnectorInventoryTranslatorImpl flowNodeConnectorInventoryTranslatorImpl;
  90     private DeviceMastershipManager deviceMastershipManager;
  91     private final ReconciliationManager reconciliationManager;
  92     private DevicesGroupRegistry devicesGroupRegistry;
  93     private NodeConfigurator nodeConfigurator;
  94     private ArbitratorReconcileService arbitratorReconciliationManager;
  95     private boolean disableReconciliation;
  96     private boolean staleMarkingEnabled;
  97     private int reconciliationRetryCount;
  98     private boolean isBundleBasedReconciliationEnabled;
  99     private final OpenflowServiceRecoveryHandler openflowServiceRecoveryHandler;
 100     private final ServiceRecoveryRegistry serviceRecoveryRegistry;
 101
 102     public ForwardingRulesManagerImpl(final DataBroker dataBroker, final RpcProviderRegistry rpcRegistry,
 103                                       final ForwardingRulesManagerConfig config,
 104                                       final MastershipChangeServiceManager mastershipChangeServiceManager,
 105                                       final ClusterSingletonServiceProvider clusterSingletonService,
 106                                       final ConfigurationService configurationService,
 107                                       final ReconciliationManager reconciliationManager,
 108                                       final OpenflowServiceRecoveryHandler openflowServiceRecoveryHandler,
 109                                       final ServiceRecoveryRegistry serviceRecoveryRegistry) {
 110         disableReconciliation = config.isDisableReconciliation();
 111         staleMarkingEnabled = config.isStaleMarkingEnabled();
 112         reconciliationRetryCount = config.getReconciliationRetryCount();
 113         isBundleBasedReconciliationEnabled = config.isBundleBasedReconciliationEnabled();
 114         this.configurationServiceRegistration = configurationService.registerListener(this);
 115         this.dataService = Preconditions.checkNotNull(dataBroker, "DataBroker can not be null!");
 116         this.clusterSingletonServiceProvider = Preconditions.checkNotNull(clusterSingletonService,
 117                 "ClusterSingletonService provider can not be null");
 118         this.reconciliationManager = reconciliationManager;
 119         this.rpcRegistry = rpcRegistry;
 120         this.mastershipChangeServiceManager = mastershipChangeServiceManager;
 121
 122         Preconditions.checkArgument(rpcRegistry != null, "RpcProviderRegistry can not be null !");
 123
 124         this.salFlowService = Preconditions.checkNotNull(rpcRegistry.getRpcService(SalFlowService.class),
 125                 "RPC SalFlowService not found.");
 126         this.salGroupService = Preconditions.checkNotNull(rpcRegistry.getRpcService(SalGroupService.class),
 127                 "RPC SalGroupService not found.");
 128         this.salMeterService = Preconditions.checkNotNull(rpcRegistry.getRpcService(SalMeterService.class),
 129                 "RPC SalMeterService not found.");
 130         this.salTableService = Preconditions.checkNotNull(rpcRegistry.getRpcService(SalTableService.class),
 131                 "RPC SalTableService not found.");
 132         this.salBundleService = Preconditions.checkNotNull(rpcRegistry.getRpcService(SalBundleService.class),
 133                 "RPC SalBundlService not found.");
 134         this.openflowServiceRecoveryHandler = Preconditions.checkNotNull(openflowServiceRecoveryHandler,
 135                 "Openflow service recovery handler cannot be null");
 136         this.serviceRecoveryRegistry = Preconditions.checkNotNull(serviceRecoveryRegistry,
 137                 "Service recovery registry cannot be null");
 138         this.arbitratorReconciliationManager = Preconditions
 139                 .checkNotNull(rpcRegistry.getRpcService(ArbitratorReconcileService.class),
 140                         "ArbitratorReconciliationManager can not be null!");
 141     }
 142
 143     @Override
 144     public void start() {
 145         nodeConfigurator = new NodeConfiguratorImpl();
 146         this.devicesGroupRegistry = new DevicesGroupRegistry();
 147
 148         this.nodeListener = new FlowNodeReconciliationImpl(this, dataService, SERVICE_NAME, FRM_RECONCILIATION_PRIORITY,
 149                 ResultState.DONOTHING);
 150         if (this.isReconciliationDisabled()) {
 151             LOG.debug("Reconciliation is disabled by user");
 152         } else {
 153             this.reconciliationNotificationRegistration = reconciliationManager.registerService(this.nodeListener);
 154             LOG.debug("Reconciliation is enabled by user and successfully registered to the reconciliation framework");
 155         }
 156         this.deviceMastershipManager = new DeviceMastershipManager(clusterSingletonServiceProvider, this.nodeListener,
 157                 dataService, mastershipChangeServiceManager);
 158         this.deviceMastershipManager.setRoutedRpcReg(rpcRegistry.addRoutedRpcImplementation(
 159                 FrmReconciliationService.class, new FrmReconciliationServiceImpl(this)));
 160         flowNodeConnectorInventoryTranslatorImpl = new FlowNodeConnectorInventoryTranslatorImpl(dataService);
 161
 162         this.flowListener = new FlowForwarder(this, dataService);
 163         this.groupListener = new GroupForwarder(this, dataService);
 164         this.meterListener = new MeterForwarder(this, dataService);
 165         this.tableListener = new TableForwarder(this, dataService);
 166         LOG.info("ForwardingRulesManager has started successfully.");
 167     }
 168
 169     @Override
 170     public void close() throws Exception {
 171         configurationServiceRegistration.close();
 172
 173         if (this.flowListener != null) {
 174             this.flowListener.close();
 175             this.flowListener = null;
 176         }
 177         if (this.groupListener != null) {
 178             this.groupListener.close();
 179             this.groupListener = null;
 180         }
 181         if (this.meterListener != null) {
 182             this.meterListener.close();
 183             this.meterListener = null;
 184         }
 185         if (this.tableListener != null) {
 186             this.tableListener.close();
 187             this.tableListener = null;
 188         }
 189         if (this.nodeListener != null) {
 190             this.nodeListener.close();
 191             this.nodeListener = null;
 192         }
 193         if (deviceMastershipManager != null) {
 194             deviceMastershipManager.close();
 195         }
 196         if (this.reconciliationNotificationRegistration != null) {
 197             this.reconciliationNotificationRegistration.close();
 198             this.reconciliationNotificationRegistration = null;
 199         }
 200     }
 201
 202     @Override
 203     public ReadOnlyTransaction getReadTransaction() {
 204         return dataService.newReadOnlyTransaction();
 205     }
 206
 207     @Override
 208     public String getNewTransactionId() {
 209         return "DOM-" + txNum.getAndIncrement();
 210     }
 211
 212     @Override
 213     public boolean isNodeActive(InstanceIdentifier<FlowCapableNode> ident) {
 214         return deviceMastershipManager.isNodeActive(ident.firstKeyOf(Node.class).getId());
 215     }
 216
 217     @Override
 218     public boolean checkNodeInOperationalDataStore(InstanceIdentifier<FlowCapableNode> ident) {
 219         boolean result = false;
 220         InstanceIdentifier<Node> nodeIid = ident.firstIdentifierOf(Node.class);
 221         final ReadOnlyTransaction transaction = dataService.newReadOnlyTransaction();
 222         ListenableFuture<com.google.common.base.Optional<Node>> future = transaction
 223                 .read(LogicalDatastoreType.OPERATIONAL, nodeIid);
 224         try {
 225             com.google.common.base.Optional<Node> optionalDataObject = future.get();
 226             if (optionalDataObject.isPresent()) {
 227                 result = true;
 228             } else {
 229                 LOG.debug("{}: Failed to read {}", Thread.currentThread().getStackTrace()[1], nodeIid);
 230             }
 231         } catch (ExecutionException | InterruptedException e) {
 232             LOG.warn("Failed to read {} ", nodeIid, e);
 233         }
 234         transaction.close();
 235
 236         return result;
 237     }
 238
 239     @Override
 240     public SalFlowService getSalFlowService() {
 241         return salFlowService;
 242     }
 243
 244     @Override
 245     public SalGroupService getSalGroupService() {
 246         return salGroupService;
 247     }
 248
 249     @Override
 250     public SalMeterService getSalMeterService() {
 251         return salMeterService;
 252     }
 253
 254     @Override
 255     public SalTableService getSalTableService() {
 256         return salTableService;
 257     }
 258
 259     @Override
 260     public DevicesGroupRegistry getDevicesGroupRegistry() {
 261         return this.devicesGroupRegistry;
 262     }
 263
 264     @Override
 265     public SalBundleService getSalBundleService() {
 266         return salBundleService;
 267     }
 268
 269     @Override
 270     public ForwardingRulesCommiter<Flow> getFlowCommiter() {
 271         return flowListener;
 272     }
 273
 274     @Override
 275     public ForwardingRulesCommiter<Group> getGroupCommiter() {
 276         return groupListener;
 277     }
 278
 279     @Override
 280     public ForwardingRulesCommiter<Meter> getMeterCommiter() {
 281         return meterListener;
 282     }
 283
 284     @Override
 285     public ForwardingRulesCommiter<TableFeatures> getTableFeaturesCommiter() {
 286         return tableListener;
 287     }
 288
 289     @Override
 290     public ArbitratorReconcileService getArbitratorReconciliationManager() {
 291         return arbitratorReconciliationManager;
 292     }
 293
 294     @Override
 295     public boolean isReconciliationDisabled() {
 296         return disableReconciliation;
 297     }
 298
 299     @Override
 300     public boolean isStaleMarkingEnabled() {
 301         return staleMarkingEnabled;
 302     }
 303
 304     @Override
 305     public int getReconciliationRetryCount() {
 306         return reconciliationRetryCount;
 307     }
 308
 309     @Override
 310     public void addRecoverableListener(RecoverableListener recoverableListener) {
 311         serviceRecoveryRegistry.addRecoverableListener(openflowServiceRecoveryHandler.buildServiceRegistryKey(),
 312                 recoverableListener);
 313     }
 314
 315     @Override
 316     public FlowNodeConnectorInventoryTranslatorImpl getFlowNodeConnectorInventoryTranslatorImpl() {
 317         return flowNodeConnectorInventoryTranslatorImpl;
 318     }
 319
 320     @Override
 321     public NodeConfigurator getNodeConfigurator() {
 322         return nodeConfigurator;
 323     }
 324
 325     public FlowNodeReconciliation getNodeListener() {
 326         return nodeListener;
 327     }
 328
 329     @Override
 330     public boolean isBundleBasedReconciliationEnabled() {
 331         return isBundleBasedReconciliationEnabled;
 332     }
 333
 334     @Override
 335     public boolean isNodeOwner(InstanceIdentifier<FlowCapableNode> ident) {
 336         return Objects.nonNull(ident) && deviceMastershipManager.isDeviceMastered(ident.firstKeyOf(Node.class).getId());
 337     }
 338
 339     @VisibleForTesting
 340     public void setDeviceMastershipManager(final DeviceMastershipManager deviceMastershipManager) {
 341         this.deviceMastershipManager = deviceMastershipManager;
 342     }
 343
 344     @Override
 345     public void onPropertyChanged(@Nonnull final String propertyName, @Nonnull final String propertyValue) {
 346         Optional.ofNullable(ForwardingRulesProperty.forValue(propertyName)).ifPresent(forwardingRulesProperty -> {
 347             switch (forwardingRulesProperty) {
 348                 case DISABLE_RECONCILIATION:
 349                     disableReconciliation = Boolean.valueOf(propertyValue);
 350                     break;
 351                 case STALE_MARKING_ENABLED:
 352                     staleMarkingEnabled = Boolean.valueOf(propertyValue);
 353                     break;
 354                 case RECONCILIATION_RETRY_COUNT:
 355                     reconciliationRetryCount = Integer.parseInt(propertyValue);
 356                     break;
 357                 case BUNDLE_BASED_RECONCILIATION_ENABLED:
 358                     isBundleBasedReconciliationEnabled = Boolean.valueOf(propertyValue);
 359                     break;
 360                 default:
 361                     LOG.warn("No forwarding rule property found.");
 362                     break;
 363             }
 364         });
 365     }
 366 }