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