openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/LifecycleConductorImpl.java

   1 /*
   2  * Copyright (c) 2016 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.impl;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import io.netty.util.HashedWheelTimer;
  18 import io.netty.util.Timeout;
  19 import io.netty.util.TimerTask;
  20 import java.util.Collection;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import java.util.concurrent.TimeUnit;
  25 import javax.annotation.Nonnull;
  26 import javax.annotation.Nullable;
  27 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  28 import org.opendaylight.openflowplugin.api.openflow.OFPManager;
  29 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  30 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  31 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  32 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  33 import org.opendaylight.openflowplugin.api.openflow.lifecycle.DeviceContextChangeListener;
  34 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
  35 import org.opendaylight.openflowplugin.api.openflow.lifecycle.RoleChangeListener;
  36 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ServiceChangeListener;
  37 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  38 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  39 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
  40 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  41 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  42 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
  43 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  44 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  45 import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
  46 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51
  52 /**
  53  */
  54 final class LifecycleConductorImpl implements LifecycleConductor, RoleChangeListener, DeviceContextChangeListener, ExtensionConverterProviderKeeper {
  55
  56     private static final Logger LOG = LoggerFactory.getLogger(LifecycleConductorImpl.class);
  57     private static final int TICKS_PER_WHEEL = 500;
  58     private static final long TICK_DURATION = 10; // 0.5 sec.
  59
  60     private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, TICKS_PER_WHEEL);
  61     private ExtensionConverterProvider extensionConverterProvider;
  62     private DeviceManager deviceManager;
  63     private StatisticsManager statisticsManager;
  64     private RpcManager rpcManager;
  65     private final MessageIntelligenceAgency messageIntelligenceAgency;
  66     private final ConvertorExecutor convertorExecutor;
  67     private ConcurrentHashMap<DeviceInfo, ServiceChangeListener> serviceChangeListeners = new ConcurrentHashMap<>();
  68     private NotificationPublishService notificationPublishService;
  69
  70     LifecycleConductorImpl(final MessageIntelligenceAgency messageIntelligenceAgency, ConvertorExecutor convertorExecutor) {
  71         this.messageIntelligenceAgency = Preconditions.checkNotNull(messageIntelligenceAgency);
  72         this.convertorExecutor = convertorExecutor;
  73     }
  74
  75     @Override
  76     public ExtensionConverterProvider getExtensionConverterProvider() {
  77         return extensionConverterProvider;
  78     }
  79
  80     @Override
  81     public void setExtensionConverterProvider(ExtensionConverterProvider extensionConverterProvider) {
  82         this.extensionConverterProvider = extensionConverterProvider;
  83     }
  84
  85     @Override
  86     public void setSafelyManager(final OFPManager manager){
  87         if (manager instanceof RpcManager) {
  88             if (rpcManager != null) {
  89                 LOG.info("RPC manager {} is already defined in conductor. ", manager);
  90                 return;
  91             }
  92             this.rpcManager = (RpcManager) manager;
  93         } else {
  94             if (manager instanceof StatisticsManager) {
  95                 if (statisticsManager != null) {
  96                     LOG.info("Statistics manager {} is already defined in conductor. ", manager);
  97                     return;
  98                 }
  99                 this.statisticsManager = (StatisticsManager) manager;
 100             } else {
 101                 if (manager instanceof DeviceManager) {
 102                     if (deviceManager != null) {
 103                         LOG.info("Device manager {} is already defined in conductor. ", manager);
 104                         return;
 105                     }
 106                     this.deviceManager = (DeviceManager) manager;
 107                 }
 108             }
 109         }
 110     }
 111
 112     @Override
 113     public void addOneTimeListenerWhenServicesChangesDone(final ServiceChangeListener manager, final DeviceInfo deviceInfo){
 114         LOG.debug("Listener {} for service change for node {} registered.", manager, deviceInfo.getNodeId());
 115         serviceChangeListeners.put(deviceInfo, manager);
 116     }
 117
 118     @VisibleForTesting
 119     void notifyServiceChangeListeners(final DeviceInfo deviceInfo, final boolean success){
 120         if (serviceChangeListeners.size() == 0) {
 121             return;
 122         }
 123         LOG.debug("Notifying registered listeners for service change, no. of listeners {}", serviceChangeListeners.size());
 124         for (final Map.Entry<DeviceInfo, ServiceChangeListener> nodeIdServiceChangeListenerEntry : serviceChangeListeners.entrySet()) {
 125             if (nodeIdServiceChangeListenerEntry.getKey().equals(deviceInfo)) {
 126                 LOG.debug("Listener {} for service change for node {} was notified. Success was set on {}", nodeIdServiceChangeListenerEntry.getValue(), deviceInfo.getNodeId().getValue(), success);
 127                 nodeIdServiceChangeListenerEntry.getValue().servicesChangeDone(deviceInfo, success);
 128                 serviceChangeListeners.remove(deviceInfo);
 129             }
 130         }
 131     }
 132
 133     @Override
 134     public void roleInitializationDone(final DeviceInfo deviceInfo, final boolean success) {
 135         if (!success) {
 136             LOG.warn("Initialization phase for node {} in role context was NOT successful, closing connection.", deviceInfo.getNodeId().getValue());
 137             closeConnection(deviceInfo);
 138         } else {
 139             LOG.info("initialization phase for node {} in role context was successful, continuing to next context.", deviceInfo.getNodeId().getValue());
 140         }
 141     }
 142
 143     public void closeConnection(final DeviceInfo deviceInfo) {
 144         LOG.debug("Close connection called for node {}", deviceInfo);
 145         final DeviceContext deviceContext = getDeviceContext(deviceInfo);
 146         if (null != deviceContext) {
 147             deviceContext.shutdownConnection();
 148         }
 149     }
 150
 151     @Override
 152     public void roleChangeOnDevice(final DeviceInfo deviceInfo, final OfpRole newRole) {
 153
 154         final DeviceContext deviceContext = Preconditions.checkNotNull(
 155                 deviceManager.gainContext(deviceInfo),
 156                 "Something went wrong, device context for nodeId: %s doesn't exists", deviceInfo.getNodeId().getValue()
 157         );
 158
 159         final RpcContext rpcContext =  Preconditions.checkNotNull(
 160                 rpcManager.gainContext(deviceInfo),
 161                 "Something went wrong, rpc context for nodeId: %s doesn't exists", deviceInfo.getNodeId().getValue()
 162         );
 163
 164         LOG.info("Role change to {} in role context for node {} was successful.", newRole, deviceInfo.getNodeId().getValue());
 165
 166         if (OfpRole.BECOMEMASTER.equals(newRole)) {
 167             fillDeviceFlowRegistry(deviceInfo, deviceContext.getDeviceFlowRegistry());
 168             MdSalRegistrationUtils.registerServices(rpcContext, deviceContext, this.extensionConverterProvider, convertorExecutor);
 169
 170             if (rpcContext.isStatisticsRpcEnabled()) {
 171                 MdSalRegistrationUtils.registerStatCompatibilityServices(
 172                         rpcContext,
 173                         deviceContext,
 174                         notificationPublishService, convertorExecutor);
 175             }
 176         } else {
 177             statisticsManager.stopScheduling(deviceInfo);
 178
 179             // Clean device flow registry if we became slave
 180             if (OfpRole.BECOMESLAVE.equals(newRole)) {
 181                 deviceContext.getDeviceFlowRegistry().close();
 182             }
 183
 184             MdSalRegistrationUtils.unregisterServices(rpcContext);
 185         }
 186
 187     }
 188
 189     private void fillDeviceFlowRegistry(final DeviceInfo deviceInfo, final DeviceFlowRegistry deviceFlowRegistry) {
 190         // Fill device flow registry with flows from datastore
 191         final ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill = deviceFlowRegistry.fill();
 192
 193         // Start statistics scheduling only after we finished initializing device flow registry
 194         Futures.addCallback(deviceFlowRegistryFill, new FutureCallback<List<Optional<FlowCapableNode>>>() {
 195             @Override
 196             public void onSuccess(@Nullable List<Optional<FlowCapableNode>> result) {
 197                 if (LOG.isDebugEnabled()) {
 198                     // Count all flows we read from datastore for debugging purposes.
 199                     // This number do not always represent how many flows were actually added
 200                     // to DeviceFlowRegistry, because of possible duplicates.
 201                     long flowCount = Optional.fromNullable(result).asSet().stream()
 202                             .flatMap(Collection::stream)
 203                             .flatMap(flowCapableNodeOptional -> flowCapableNodeOptional.asSet().stream())
 204                             .flatMap(flowCapableNode -> flowCapableNode.getTable().stream())
 205                             .flatMap(table -> table.getFlow().stream())
 206                             .count();
 207
 208                     LOG.debug("Finished filling flow registry with {} flows for node: {}", flowCount, deviceInfo.getNodeId().getValue());
 209                 }
 210
 211                 statisticsManager.startScheduling(deviceInfo);
 212             }
 213
 214             @Override
 215             public void onFailure(Throwable t) {
 216                 // If we manually cancelled this future, do not start scheduling of statistics
 217                 if (deviceFlowRegistryFill.isCancelled()) {
 218                     LOG.debug("Cancelled filling flow registry with flows for node: {}", deviceInfo.getNodeId().getValue());
 219                 } else {
 220                     LOG.warn("Failed filling flow registry with flows for node: {} with exception: {}", deviceInfo.getNodeId().getValue(), t);
 221                     statisticsManager.startScheduling(deviceInfo);
 222                 }
 223             }
 224         });
 225     }
 226
 227     public MessageIntelligenceAgency getMessageIntelligenceAgency() {
 228         return messageIntelligenceAgency;
 229     }
 230
 231     @Override
 232     public DeviceContext getDeviceContext(DeviceInfo deviceInfo){
 233          return deviceManager.gainContext(deviceInfo);
 234     }
 235
 236     @Override
 237     public StatisticsContext getStatisticsContext(DeviceInfo deviceInfo){
 238         return statisticsManager.gainContext(deviceInfo);
 239     }
 240
 241     public Timeout newTimeout(@Nonnull TimerTask task, long delay, @Nonnull TimeUnit unit) {
 242         return hashedWheelTimer.newTimeout(task, delay, unit);
 243     }
 244
 245     @Override
 246     public ConnectionContext.CONNECTION_STATE gainConnectionStateSafely(final DeviceInfo deviceInfo){
 247         return (null != getDeviceContext(deviceInfo)) ? getDeviceContext(deviceInfo).getPrimaryConnectionContext().getConnectionState() : null;
 248     }
 249
 250     @Override
 251     public Long reserveXidForDeviceMessage(final DeviceInfo deviceInfo){
 252         return null != getDeviceContext(deviceInfo) ? getDeviceContext(deviceInfo).reserveXidForDeviceMessage() : null;
 253     }
 254
 255     @Override
 256     public void deviceStartInitializationDone(final DeviceInfo deviceInfo, final boolean success) {
 257         if (!success) {
 258             LOG.warn("Initialization phase for node {} in device context was NOT successful, closing connection.", deviceInfo.getNodeId().getValue());
 259             closeConnection(deviceInfo);
 260         } else {
 261             LOG.info("initialization phase for node {} in device context was successful. Continuing to next context.", deviceInfo.getNodeId().getValue());
 262         }
 263     }
 264
 265     @Override
 266     public void deviceInitializationDone(final DeviceInfo deviceInfo, final boolean success) {
 267         if (!success) {
 268             LOG.warn("Initialization phase for node {} in device context was NOT successful, closing connection.", deviceInfo.getNodeId().getValue());
 269             closeConnection(deviceInfo);
 270         } else {
 271             LOG.info("initialization phase for node {} in device context was successful. All phases initialized OK.", deviceInfo.getNodeId().getValue());
 272         }
 273     }
 274
 275     @VisibleForTesting
 276     boolean isServiceChangeListenersEmpty() {
 277         return this.serviceChangeListeners.isEmpty();
 278     }
 279
 280     @Override
 281     public NotificationPublishService getNotificationPublishService() {
 282         return notificationPublishService;
 283     }
 284
 285     @Override
 286     public void setNotificationPublishService(NotificationPublishService notificationPublishService) {
 287         this.notificationPublishService = notificationPublishService;
 288     }
 289 }