2 * Copyright (c) 2016 Pantheon Technologies s.r.o. and others. All rights reserved.
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
8 package org.opendaylight.openflowplugin.impl.lifecycle;
10 import com.google.common.annotations.VisibleForTesting;
11 import com.google.common.util.concurrent.FutureCallback;
12 import com.google.common.util.concurrent.Futures;
13 import com.google.common.util.concurrent.ListenableFuture;
14 import com.google.common.util.concurrent.MoreExecutors;
15 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
16 import java.util.HashMap;
17 import java.util.List;
19 import java.util.Objects;
20 import java.util.Optional;
21 import java.util.concurrent.ConcurrentHashMap;
22 import java.util.concurrent.ExecutionException;
23 import java.util.concurrent.ExecutorService;
24 import java.util.concurrent.TimeUnit;
25 import java.util.concurrent.TimeoutException;
26 import java.util.stream.Collectors;
27 import javax.annotation.Nonnull;
28 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipChange;
29 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipListenerRegistration;
30 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipService;
31 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
32 import org.opendaylight.openflowplugin.api.openflow.OFPManager;
33 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
34 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionStatus;
35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
38 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
39 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
40 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
41 import org.opendaylight.openflowplugin.api.openflow.lifecycle.MasterChecker;
42 import org.opendaylight.openflowplugin.api.openflow.lifecycle.OwnershipChangeListener;
43 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
44 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
45 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
46 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
47 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
48 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
49 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.mdsal.core.general.entity.rev150930.Entity;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
56 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
57 import org.opendaylight.yangtools.yang.common.Uint8;
58 import org.slf4j.Logger;
59 import org.slf4j.LoggerFactory;
61 public class ContextChainHolderImpl implements ContextChainHolder, MasterChecker {
62 private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
63 private static final Logger OF_EVENT_LOG = LoggerFactory.getLogger("OfEventLog");
65 private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
66 private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
67 private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
68 private static final String SEPARATOR = ":";
69 private final Map<DeviceInfo, ContextChain> contextChainMap = new ConcurrentHashMap<>();
70 private final Map<DeviceInfo, ? super ConnectionContext> connectingDevices = new ConcurrentHashMap<>();
71 private final EntityOwnershipListenerRegistration eosListenerRegistration;
72 private final ClusterSingletonServiceProvider singletonServiceProvider;
73 private final ExecutorService executorService;
74 private final OwnershipChangeListener ownershipChangeListener;
75 private DeviceManager deviceManager;
76 private RpcManager rpcManager;
77 private StatisticsManager statisticsManager;
78 private RoleManager roleManager;
80 public ContextChainHolderImpl(final ExecutorService executorService,
81 final ClusterSingletonServiceProvider singletonServiceProvider,
82 final EntityOwnershipService entityOwnershipService,
83 final OwnershipChangeListener ownershipChangeListener) {
84 this.singletonServiceProvider = singletonServiceProvider;
85 this.executorService = executorService;
86 this.ownershipChangeListener = ownershipChangeListener;
87 this.ownershipChangeListener.setMasterChecker(this);
88 this.eosListenerRegistration = Objects
89 .requireNonNull(entityOwnershipService.registerListener(ASYNC_SERVICE_ENTITY_TYPE, this));
93 public <T extends OFPManager> void addManager(final T manager) {
94 if (deviceManager == null && manager instanceof DeviceManager) {
95 LOG.trace("Context chain holder: Device manager OK.");
96 deviceManager = (DeviceManager) manager;
97 } else if (rpcManager == null && manager instanceof RpcManager) {
98 LOG.trace("Context chain holder: RPC manager OK.");
99 rpcManager = (RpcManager) manager;
100 } else if (statisticsManager == null && manager instanceof StatisticsManager) {
101 LOG.trace("Context chain holder: Statistics manager OK.");
102 statisticsManager = (StatisticsManager) manager;
103 } else if (roleManager == null && manager instanceof RoleManager) {
104 LOG.trace("Context chain holder: Role manager OK.");
105 roleManager = (RoleManager) manager;
110 void createContextChain(final ConnectionContext connectionContext) {
111 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
113 final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
114 deviceContext.registerMastershipWatcher(this);
115 LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
117 final RpcContext rpcContext = rpcManager.createContext(deviceContext);
118 rpcContext.registerMastershipWatcher(this);
119 LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
121 final StatisticsContext statisticsContext = statisticsManager
122 .createContext(deviceContext, ownershipChangeListener.isReconciliationFrameworkRegistered());
123 statisticsContext.registerMastershipWatcher(this);
124 LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
126 final RoleContext roleContext = roleManager.createContext(deviceContext);
127 roleContext.registerMastershipWatcher(this);
128 LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
130 final ContextChain contextChain = new ContextChainImpl(this, connectionContext, executorService);
131 contextChain.registerDeviceRemovedHandler(deviceManager);
132 contextChain.registerDeviceRemovedHandler(rpcManager);
133 contextChain.registerDeviceRemovedHandler(statisticsManager);
134 contextChain.registerDeviceRemovedHandler(roleManager);
135 contextChain.registerDeviceRemovedHandler(this);
136 contextChain.addContext(deviceContext);
137 contextChain.addContext(rpcContext);
138 contextChain.addContext(statisticsContext);
139 contextChain.addContext(roleContext);
140 contextChainMap.put(deviceInfo, contextChain);
141 connectingDevices.remove(deviceInfo);
142 LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
144 deviceContext.onPublished();
145 contextChain.registerServices(singletonServiceProvider);
149 public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) {
150 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
151 final ContextChain contextChain = contextChainMap.get(deviceInfo);
152 final FeaturesReply featuresReply = connectionContext.getFeatures();
153 final Uint8 auxiliaryId = featuresReply != null ? featuresReply.getAuxiliaryId() : null;
155 if (auxiliaryId != null && auxiliaryId.toJava() != 0) {
156 if (contextChain == null) {
157 LOG.warn("An auxiliary connection for device {}, but no primary connection. Refusing connection.",
159 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
161 if (contextChain.addAuxiliaryConnection(connectionContext)) {
162 LOG.info("An auxiliary connection was added to device: {}", deviceInfo);
163 return ConnectionStatus.MAY_CONTINUE;
165 LOG.warn("Not able to add auxiliary connection to the device {}", deviceInfo);
166 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
170 LOG.info("Device {} connected.", deviceInfo);
171 final boolean contextExists = contextChain != null;
172 final boolean isClosing = contextExists && contextChain.isClosing();
174 if (!isClosing && connectingDevices.putIfAbsent(deviceInfo, connectionContext) != null) {
175 LOG.warn("Device {} is already trying to connect, wait until succeeded or disconnected.", deviceInfo);
176 return ConnectionStatus.ALREADY_CONNECTED;
181 LOG.warn("Device {} is already in termination state, closing all incoming connections.",
183 return ConnectionStatus.CLOSING;
186 LOG.warn("Device {} already connected. Closing previous connection", deviceInfo);
187 destroyContextChain(deviceInfo);
188 LOG.info("Old connection dropped, creating new context chain for device {}", deviceInfo);
189 createContextChain(connectionContext);
191 LOG.info("No context chain found for device: {}, creating new.", deviceInfo);
192 createContextChain(connectionContext);
195 return ConnectionStatus.MAY_CONTINUE;
201 public void onNotAbleToStartMastership(@Nonnull final DeviceInfo deviceInfo, @Nonnull final String reason,
202 final boolean mandatory) {
203 LOG.warn("Not able to set MASTER role on device {}, reason: {}", deviceInfo, reason);
209 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
210 LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.",
212 destroyContextChain(deviceInfo);
217 public void onMasterRoleAcquired(@Nonnull final DeviceInfo deviceInfo,
218 @Nonnull final ContextChainMastershipState mastershipState) {
219 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
220 if (ownershipChangeListener.isReconciliationFrameworkRegistered()
221 && !ContextChainMastershipState.INITIAL_SUBMIT.equals(mastershipState)) {
222 if (contextChain.isMastered(mastershipState, true)) {
223 Futures.addCallback(ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
224 reconciliationFrameworkCallback(deviceInfo, contextChain, mastershipState),
225 MoreExecutors.directExecutor());
227 } else if (contextChain.isMastered(mastershipState, false)) {
228 LOG.info("Role MASTER was granted to device {}", deviceInfo);
229 ownershipChangeListener.becomeMaster(deviceInfo);
230 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
236 public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
237 ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
238 LOG.info("Role SLAVE was granted to device {}", deviceInfo);
239 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
243 public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
244 LOG.warn("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
245 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
249 public void onDeviceDisconnected(final ConnectionContext connectionContext) {
250 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
252 Optional.ofNullable(connectionContext.getDeviceInfo()).map(contextChainMap::get).ifPresent(contextChain -> {
253 if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
254 LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
256 LOG.info("Device {} disconnected.", deviceInfo);
257 destroyContextChain(deviceInfo);
263 boolean checkAllManagers() {
264 return deviceManager != null && rpcManager != null && statisticsManager != null && roleManager != null;
268 public ContextChain getContextChain(final DeviceInfo deviceInfo) {
269 return contextChainMap.get(deviceInfo);
273 public void close() {
274 Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
275 copyOfChains.keySet().forEach(this::destroyContextChain);
276 copyOfChains.clear();
277 contextChainMap.clear();
278 eosListenerRegistration.close();
279 OF_EVENT_LOG.debug("EOS registration closed for all devices");
283 @SuppressFBWarnings("BC_UNCONFIRMED_CAST_OF_RETURN_VALUE")
284 public void ownershipChanged(final EntityOwnershipChange entityOwnershipChange) {
285 if (entityOwnershipChange.getState().hasOwner()) {
289 // Findbugs flags a false violation for "Unchecked/unconfirmed cast" from GenericEntity to Entity hence the
290 // suppression above. The suppression is temporary until EntityOwnershipChange is modified to eliminate the
292 final String entityName = entityOwnershipChange
295 .firstKeyOf(Entity.class)
298 if (entityName != null) {
299 LOG.debug("Entity {} has no owner", entityName);
300 final String dpnId = getDpnIdFromNodeName(entityName);
302 //TODO:Remove notifications
303 final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
304 DeviceStateUtil.createNodeInstanceIdentifier(new NodeId(entityName));
305 deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
306 LOG.info("Try to remove device {} from operational DS", entityName);
307 ListenableFuture<?> future = deviceManager.removeDeviceFromOperationalDS(nodeInstanceIdentifier);
308 Futures.addCallback(future, new FutureCallback<Object>() {
310 public void onSuccess(final Object result) {
311 LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
312 OF_EVENT_LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
316 public void onFailure(final Throwable throwable) {
317 LOG.error("Could not remove device {} from operational DS", dpnId, throwable);
319 }, MoreExecutors.directExecutor());
320 future.get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
321 } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
322 LOG.warn("Not able to remove device {} from operational DS. ", entityName, e);
324 OF_EVENT_LOG.debug("Node removed, Node: {}", dpnId);
325 LOG.info("Removing device from operational DS {} was successful", dpnId);
329 private void destroyContextChain(final DeviceInfo deviceInfo) {
330 OF_EVENT_LOG.debug("Destroying context chain for device {}", deviceInfo.getDatapathId());
331 ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
332 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
333 deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
334 contextChain.close();
335 connectingDevices.remove(deviceInfo);
340 public List<DeviceInfo> listOfMasteredDevices() {
341 return contextChainMap.entrySet().stream()
342 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
343 .isMastered(ContextChainMastershipState.CHECK, false)).map(Map.Entry::getKey)
344 .collect(Collectors.toList());
348 public boolean isAnyDeviceMastered() {
349 return contextChainMap.entrySet().stream().findAny()
350 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
351 .isMastered(ContextChainMastershipState.CHECK, false)).isPresent();
355 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
356 contextChainMap.remove(deviceInfo);
357 LOG.debug("Context chain removed for node {}", deviceInfo);
360 private FutureCallback<ResultState> reconciliationFrameworkCallback(@Nonnull final DeviceInfo deviceInfo,
361 final ContextChain contextChain, final ContextChainMastershipState mastershipState) {
362 return new FutureCallback<>() {
364 public void onSuccess(final ResultState result) {
365 if (ResultState.DONOTHING == result) {
366 OF_EVENT_LOG.debug("Device {} connection is enabled by reconciliation framework", deviceInfo);
367 LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
368 if (mastershipState == ContextChainMastershipState.MASTER_ON_DEVICE) {
369 contextChain.initializeDevice();
371 contextChain.continueInitializationAfterReconciliation();
373 OF_EVENT_LOG.debug("Reconciliation framework failure for device {}", deviceInfo);
374 LOG.warn("Reconciliation framework failure for device {}", deviceInfo);
375 destroyContextChain(deviceInfo);
380 public void onFailure(final Throwable throwable) {
381 OF_EVENT_LOG.debug("Reconciliation framework failure for device {} with error {}", deviceInfo,
382 throwable.getMessage());
383 LOG.warn("Reconciliation framework failure.");
384 destroyContextChain(deviceInfo);
389 private String getDpnIdFromNodeName(final String nodeName) {
390 return nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1);