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.MoreExecutors;
14 import java.util.HashMap;
15 import java.util.List;
17 import java.util.Objects;
18 import java.util.Optional;
19 import java.util.concurrent.ConcurrentHashMap;
20 import java.util.concurrent.ExecutionException;
21 import java.util.concurrent.ExecutorService;
22 import java.util.concurrent.TimeUnit;
23 import java.util.concurrent.TimeoutException;
24 import java.util.stream.Collectors;
25 import javax.annotation.Nonnull;
26 import javax.annotation.Nullable;
27 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
28 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListenerRegistration;
29 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
30 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
31 import org.opendaylight.openflowplugin.api.openflow.OFPManager;
32 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
33 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionStatus;
34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
37 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
38 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
39 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
40 import org.opendaylight.openflowplugin.api.openflow.lifecycle.MasterChecker;
41 import org.opendaylight.openflowplugin.api.openflow.lifecycle.OwnershipChangeListener;
42 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
43 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
44 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
45 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
46 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
47 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
48 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
54 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
56 import org.slf4j.Logger;
57 import org.slf4j.LoggerFactory;
59 public class ContextChainHolderImpl implements ContextChainHolder, MasterChecker {
60 private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
62 private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
63 private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
64 private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
66 private final Map<DeviceInfo, ContextChain> contextChainMap = new ConcurrentHashMap<>();
67 private final Map<DeviceInfo, ? super ConnectionContext> connectingDevices = new ConcurrentHashMap<>();
68 private final EntityOwnershipListenerRegistration eosListenerRegistration;
69 private final ClusterSingletonServiceProvider singletonServiceProvider;
70 private final ExecutorService executorService;
71 private final OwnershipChangeListener ownershipChangeListener;
72 private DeviceManager deviceManager;
73 private RpcManager rpcManager;
74 private StatisticsManager statisticsManager;
75 private RoleManager roleManager;
77 public ContextChainHolderImpl(final ExecutorService executorService,
78 final ClusterSingletonServiceProvider singletonServiceProvider,
79 final EntityOwnershipService entityOwnershipService,
80 final OwnershipChangeListener ownershipChangeListener) {
81 this.singletonServiceProvider = singletonServiceProvider;
82 this.executorService = executorService;
83 this.ownershipChangeListener = ownershipChangeListener;
84 this.ownershipChangeListener.setMasterChecker(this);
85 this.eosListenerRegistration = Objects.requireNonNull(entityOwnershipService
86 .registerListener(ASYNC_SERVICE_ENTITY_TYPE, this));
90 public <T extends OFPManager> void addManager(final T manager) {
91 if (Objects.isNull(deviceManager) && manager instanceof DeviceManager) {
92 LOG.trace("Context chain holder: Device manager OK.");
93 deviceManager = (DeviceManager) manager;
94 } else if (Objects.isNull(rpcManager) && manager instanceof RpcManager) {
95 LOG.trace("Context chain holder: RPC manager OK.");
96 rpcManager = (RpcManager) manager;
97 } else if (Objects.isNull(statisticsManager) && manager instanceof StatisticsManager) {
98 LOG.trace("Context chain holder: Statistics manager OK.");
99 statisticsManager = (StatisticsManager) manager;
100 } else if (Objects.isNull(roleManager) && manager instanceof RoleManager) {
101 LOG.trace("Context chain holder: Role manager OK.");
102 roleManager = (RoleManager) manager;
107 void createContextChain(final ConnectionContext connectionContext) {
108 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
110 final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
111 deviceContext.registerMastershipWatcher(this);
112 LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
114 final RpcContext rpcContext = rpcManager.createContext(deviceContext);
115 rpcContext.registerMastershipWatcher(this);
116 LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
118 final StatisticsContext statisticsContext = statisticsManager.createContext(deviceContext,
119 ownershipChangeListener.isReconciliationFrameworkRegistered());
120 statisticsContext.registerMastershipWatcher(this);
121 LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
123 final RoleContext roleContext = roleManager.createContext(deviceContext);
124 roleContext.registerMastershipWatcher(this);
125 LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
127 final ContextChain contextChain = new ContextChainImpl(this, connectionContext,
129 contextChain.registerDeviceRemovedHandler(deviceManager);
130 contextChain.registerDeviceRemovedHandler(rpcManager);
131 contextChain.registerDeviceRemovedHandler(statisticsManager);
132 contextChain.registerDeviceRemovedHandler(roleManager);
133 contextChain.registerDeviceRemovedHandler(this);
134 contextChain.addContext(deviceContext);
135 contextChain.addContext(rpcContext);
136 contextChain.addContext(statisticsContext);
137 contextChain.addContext(roleContext);
138 contextChainMap.put(deviceInfo, contextChain);
139 connectingDevices.remove(deviceInfo);
140 LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
142 deviceContext.onPublished();
143 contextChain.registerServices(singletonServiceProvider);
147 public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) throws Exception {
149 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
150 final ContextChain contextChain = contextChainMap.get(deviceInfo);
152 final FeaturesReply featuresReply = connectionContext.getFeatures();
153 final Short auxiliaryId = featuresReply != null ? featuresReply.getAuxiliaryId() : null;
155 if (auxiliaryId != null && auxiliaryId != 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,
202 @Nonnull final String reason,
203 final boolean mandatory) {
204 LOG.warn("Not able to set MASTER role on device {}, reason: {}", deviceInfo, reason);
210 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
211 LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.",
213 destroyContextChain(deviceInfo);
218 public void onMasterRoleAcquired(@Nonnull final DeviceInfo deviceInfo,
219 @Nonnull final ContextChainMastershipState mastershipState) {
220 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
221 if (ownershipChangeListener.isReconciliationFrameworkRegistered() &&
222 !ContextChainMastershipState.INITIAL_SUBMIT.equals(mastershipState)) {
223 if (contextChain.isMastered(mastershipState, true)) {
225 ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
226 reconciliationFrameworkCallback(deviceInfo, contextChain),
227 MoreExecutors.directExecutor());
229 } else if (contextChain.isMastered(mastershipState, false)) {
230 LOG.info("Role MASTER was granted to device {}", deviceInfo);
231 ownershipChangeListener.becomeMaster(deviceInfo);
232 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
238 public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
239 ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
240 LOG.info("Role SLAVE was granted to device {}", deviceInfo);
241 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
245 public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
246 LOG.warn("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
247 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
251 public void onDeviceDisconnected(final ConnectionContext connectionContext) {
252 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
254 Optional.ofNullable(connectionContext.getDeviceInfo())
255 .map(contextChainMap::get)
256 .ifPresent(contextChain -> {
257 if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
258 LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
260 LOG.info("Device {} disconnected.", deviceInfo);
261 destroyContextChain(deviceInfo);
267 boolean checkAllManagers() {
268 return Objects.nonNull(deviceManager)
269 && Objects.nonNull(rpcManager)
270 && Objects.nonNull(statisticsManager)
271 && Objects.nonNull(roleManager);
275 public void close() throws Exception {
276 Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
277 copyOfChains.keySet().forEach(this::destroyContextChain);
278 copyOfChains.clear();
279 contextChainMap.clear();
280 eosListenerRegistration.close();
284 public void ownershipChanged(EntityOwnershipChange entityOwnershipChange) {
285 if (entityOwnershipChange.hasOwner()) {
289 final String entityName = getEntityNameFromOwnershipChange(entityOwnershipChange);
291 if (Objects.nonNull(entityName)) {
292 LOG.debug("Entity {} has no owner", entityName);
293 final NodeId nodeId = new NodeId(entityName);
296 final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
297 DeviceStateUtil.createNodeInstanceIdentifier(nodeId);
299 deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
301 LOG.info("Try to remove device {} from operational DS", nodeId);
303 .removeDeviceFromOperationalDS(nodeInstanceIdentifier)
304 .get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
305 LOG.info("Removing device from operational DS {} was successful", nodeId);
306 } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
307 LOG.warn("Not able to remove device {} from operational DS. ",nodeId, e);
312 private void destroyContextChain(final DeviceInfo deviceInfo) {
313 ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
314 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
315 deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
316 contextChain.close();
321 public List<DeviceInfo> listOfMasteredDevices() {
322 return contextChainMap
325 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry
327 .isMastered(ContextChainMastershipState.CHECK, false))
328 .map(Map.Entry::getKey)
329 .collect(Collectors.toList());
333 public boolean isAnyDeviceMastered() {
334 return contextChainMap
338 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
339 .isMastered(ContextChainMastershipState.CHECK, false))
343 private String getEntityNameFromOwnershipChange(final EntityOwnershipChange entityOwnershipChange) {
344 final YangInstanceIdentifier.NodeIdentifierWithPredicates lastIdArgument =
345 (YangInstanceIdentifier.NodeIdentifierWithPredicates) entityOwnershipChange
348 .getLastPathArgument();
350 return lastIdArgument
359 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
360 contextChainMap.remove(deviceInfo);
361 LOG.debug("Context chain removed for node {}", deviceInfo);
364 private FutureCallback<ResultState> reconciliationFrameworkCallback(
365 @Nonnull DeviceInfo deviceInfo,
366 ContextChain contextChain) {
367 return new FutureCallback<ResultState>() {
369 public void onSuccess(@Nullable ResultState result) {
370 if (ResultState.DONOTHING == result) {
371 LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
372 contextChain.continueInitializationAfterReconciliation();
374 LOG.warn("Reconciliation framework failure for device {}", deviceInfo);
375 destroyContextChain(deviceInfo);
380 public void onFailure(@Nonnull Throwable throwable) {
381 LOG.warn("Reconciliation framework failure.");
382 destroyContextChain(deviceInfo);