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 com.google.common.util.concurrent.ThreadFactoryBuilder;
16 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
17 import java.util.HashMap;
18 import java.util.List;
20 import java.util.Objects;
21 import java.util.Optional;
22 import java.util.concurrent.ConcurrentHashMap;
23 import java.util.concurrent.ExecutionException;
24 import java.util.concurrent.ExecutorService;
25 import java.util.concurrent.Executors;
26 import java.util.concurrent.ScheduledExecutorService;
27 import java.util.concurrent.ThreadFactory;
28 import java.util.concurrent.TimeUnit;
29 import java.util.concurrent.TimeoutException;
30 import java.util.stream.Collectors;
31 import org.eclipse.jdt.annotation.NonNull;
32 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipChange;
33 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipListenerRegistration;
34 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipService;
35 import org.opendaylight.mdsal.eos.common.api.EntityOwnershipState;
36 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
37 import org.opendaylight.openflowplugin.api.openflow.OFPManager;
38 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
39 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionStatus;
40 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
41 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
42 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
43 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
44 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
45 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
46 import org.opendaylight.openflowplugin.api.openflow.lifecycle.MasterChecker;
47 import org.opendaylight.openflowplugin.api.openflow.lifecycle.OwnershipChangeListener;
48 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
49 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
50 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
51 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
52 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
53 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
54 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.mdsal.core.general.entity.rev150930.Entity;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
62 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
63 import org.opendaylight.yangtools.yang.common.Uint8;
64 import org.slf4j.Logger;
65 import org.slf4j.LoggerFactory;
67 public class ContextChainHolderImpl implements ContextChainHolder, MasterChecker {
68 private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
69 private static final Logger OF_EVENT_LOG = LoggerFactory.getLogger("OfEventLog");
71 private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
72 private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
73 private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
74 private static final String SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.ServiceEntityType";
75 private static final String SEPARATOR = ":";
76 private final Map<DeviceInfo, ContextChain> contextChainMap = new ConcurrentHashMap<>();
77 private final Map<DeviceInfo, ? super ConnectionContext> connectingDevices = new ConcurrentHashMap<>();
78 private final EntityOwnershipListenerRegistration eosListenerRegistration;
79 private final ClusterSingletonServiceProvider singletonServiceProvider;
80 private final ExecutorService executorService;
81 private final OwnershipChangeListener ownershipChangeListener;
82 private final ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true)
83 .setNameFormat("node-cleaner-%d").setUncaughtExceptionHandler((thread, throwable) -> {
84 LOG.warn("Uncaught exception while removing node data from operational datastore.", throwable);
86 private final ScheduledExecutorService nodeCleanerExecutor = Executors.newScheduledThreadPool(
87 Runtime.getRuntime().availableProcessors() , threadFactory);
88 private final EntityOwnershipService entityOwnershipService;
89 private final OpenflowProviderConfig config;
90 private DeviceManager deviceManager;
91 private RpcManager rpcManager;
92 private StatisticsManager statisticsManager;
93 private RoleManager roleManager;
95 public ContextChainHolderImpl(final ExecutorService executorService,
96 final ClusterSingletonServiceProvider singletonServiceProvider,
97 final EntityOwnershipService entityOwnershipService,
98 final OwnershipChangeListener ownershipChangeListener,
99 final OpenflowProviderConfig config) {
100 this.singletonServiceProvider = singletonServiceProvider;
101 this.executorService = executorService;
102 this.ownershipChangeListener = ownershipChangeListener;
103 this.ownershipChangeListener.setMasterChecker(this);
104 this.entityOwnershipService = entityOwnershipService;
105 this.config = config;
106 this.eosListenerRegistration = Objects
107 .requireNonNull(entityOwnershipService.registerListener(ASYNC_SERVICE_ENTITY_TYPE, this));
111 public <T extends OFPManager> void addManager(final T manager) {
112 if (deviceManager == null && manager instanceof DeviceManager) {
113 LOG.trace("Context chain holder: Device manager OK.");
114 deviceManager = (DeviceManager) manager;
115 } else if (rpcManager == null && manager instanceof RpcManager) {
116 LOG.trace("Context chain holder: RPC manager OK.");
117 rpcManager = (RpcManager) manager;
118 } else if (statisticsManager == null && manager instanceof StatisticsManager) {
119 LOG.trace("Context chain holder: Statistics manager OK.");
120 statisticsManager = (StatisticsManager) manager;
121 } else if (roleManager == null && manager instanceof RoleManager) {
122 LOG.trace("Context chain holder: Role manager OK.");
123 roleManager = (RoleManager) manager;
128 void createContextChain(final ConnectionContext connectionContext) {
129 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
130 final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
131 deviceContext.registerMastershipWatcher(this);
132 LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
134 final RpcContext rpcContext = rpcManager.createContext(deviceContext);
135 rpcContext.registerMastershipWatcher(this);
136 LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
138 final StatisticsContext statisticsContext = statisticsManager
139 .createContext(deviceContext, ownershipChangeListener.isReconciliationFrameworkRegistered());
140 statisticsContext.registerMastershipWatcher(this);
141 LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
143 final RoleContext roleContext = roleManager.createContext(deviceContext);
144 roleContext.registerMastershipWatcher(this);
145 LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
147 final ContextChain contextChain = new ContextChainImpl(this, connectionContext, executorService);
148 contextChain.registerDeviceRemovedHandler(deviceManager);
149 contextChain.registerDeviceRemovedHandler(rpcManager);
150 contextChain.registerDeviceRemovedHandler(statisticsManager);
151 contextChain.registerDeviceRemovedHandler(roleManager);
152 contextChain.registerDeviceRemovedHandler(this);
153 contextChain.addContext(deviceContext);
154 contextChain.addContext(rpcContext);
155 contextChain.addContext(statisticsContext);
156 contextChain.addContext(roleContext);
157 contextChainMap.put(deviceInfo, contextChain);
158 connectingDevices.remove(deviceInfo);
159 LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
161 deviceContext.onPublished();
162 contextChain.registerServices(singletonServiceProvider);
166 public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) {
167 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
168 final ContextChain contextChain = contextChainMap.get(deviceInfo);
169 final FeaturesReply featuresReply = connectionContext.getFeatures();
170 final Uint8 auxiliaryId = featuresReply != null ? featuresReply.getAuxiliaryId() : null;
172 if (auxiliaryId != null && auxiliaryId.toJava() != 0) {
173 if (contextChain == null) {
174 LOG.warn("An auxiliary connection for device {}, but no primary connection. Refusing connection.",
176 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
178 if (contextChain.addAuxiliaryConnection(connectionContext)) {
179 LOG.info("An auxiliary connection was added to device: {}", deviceInfo);
180 return ConnectionStatus.MAY_CONTINUE;
182 LOG.warn("Not able to add auxiliary connection to the device {}", deviceInfo);
183 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
187 LOG.info("Device {} connected.", deviceInfo);
188 final boolean contextExists = contextChain != null;
189 final boolean isClosing = contextExists && contextChain.isClosing();
191 if (!isClosing && connectingDevices.putIfAbsent(deviceInfo, connectionContext) != null) {
192 LOG.warn("Device {} is already trying to connect, wait until succeeded or disconnected.", deviceInfo);
193 return ConnectionStatus.ALREADY_CONNECTED;
198 LOG.warn("Device {} is already in termination state, closing all incoming connections.",
200 return ConnectionStatus.CLOSING;
203 LOG.warn("Device {} already connected. Closing previous connection", deviceInfo);
204 destroyContextChain(deviceInfo);
205 LOG.info("Old connection dropped, creating new context chain for device {}", deviceInfo);
206 createContextChain(connectionContext);
208 LOG.info("No context chain found for device: {}, creating new.", deviceInfo);
209 createContextChain(connectionContext);
212 return ConnectionStatus.MAY_CONTINUE;
218 public void onNotAbleToStartMastership(@NonNull final DeviceInfo deviceInfo, @NonNull final String reason,
219 final boolean mandatory) {
220 LOG.error("Not able to set MASTER role on device {}, reason: {}", deviceInfo, reason);
226 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
227 LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.",
229 destroyContextChain(deviceInfo);
234 public void onMasterRoleAcquired(final DeviceInfo deviceInfo, final ContextChainMastershipState mastershipState) {
235 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
236 if (!ContextChainMastershipState.INITIAL_SUBMIT.equals(mastershipState)) {
237 if (contextChain.isMastered(mastershipState, true)) {
238 Futures.addCallback(ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
239 reconciliationFrameworkCallback(deviceInfo, contextChain, mastershipState),
240 MoreExecutors.directExecutor());
242 } else if (contextChain.isMastered(mastershipState, false)) {
243 LOG.info("Role MASTER was granted to device {}", deviceInfo);
244 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
250 public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
251 ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
252 LOG.info("Role SLAVE was granted to device {}", deviceInfo);
253 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
257 public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
258 LOG.error("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
259 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
263 public void onDeviceDisconnected(final ConnectionContext connectionContext) {
264 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
266 Optional.ofNullable(connectionContext.getDeviceInfo()).map(contextChainMap::get).ifPresent(contextChain -> {
267 if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
268 LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
270 LOG.info("Device {} disconnected.", deviceInfo);
271 destroyContextChain(deviceInfo);
277 boolean checkAllManagers() {
278 return deviceManager != null && rpcManager != null && statisticsManager != null && roleManager != null;
282 public ContextChain getContextChain(final DeviceInfo deviceInfo) {
283 return contextChainMap.get(deviceInfo);
287 public void close() {
288 Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
289 copyOfChains.keySet().forEach(this::destroyContextChain);
290 copyOfChains.clear();
291 contextChainMap.clear();
292 OF_EVENT_LOG.debug("EOS registration closed for all devices");
293 eosListenerRegistration.close();
294 OF_EVENT_LOG.debug("EOS registration closed for all devices");
295 nodeCleanerExecutor.shutdownNow();
299 @SuppressFBWarnings("BC_UNCONFIRMED_CAST_OF_RETURN_VALUE")
300 public void ownershipChanged(final EntityOwnershipChange entityOwnershipChange) {
301 LOG.info("Entity ownership change received for node : {}", entityOwnershipChange);
302 if (entityOwnershipChange.inJeopardy()) {
303 LOG.warn("Controller is in Jeopardy, ignore ownership change notification. {}", entityOwnershipChange);
306 if (entityOwnershipChange.getState().hasOwner()) {
310 // Findbugs flags a false violation for "Unchecked/unconfirmed cast" from GenericEntity to Entity hence the
311 // suppression above. The suppression is temporary until EntityOwnershipChange is modified to eliminate the
313 final String entityName = entityOwnershipChange
316 .firstKeyOf(Entity.class)
319 if (entityName != null && entityName.startsWith("openflow:")) {
320 if (nodeCleanerExecutor.isShutdown()) {
321 LOG.warn("Node cleaner executor thread-pool is down.");
324 LOG.debug("Device {} will be removed from datastore in {} msec, if it's not transient notification.",
325 entityName, config.getDeviceDatastoreRemovalDelay().getValue());
326 final String dpnId = getDpnIdFromNodeName(entityName);
327 nodeCleanerExecutor.schedule(() -> {
329 Optional<EntityOwnershipState> ownershipState = getCurrentOwnershipStatus(entityName);
330 if (!ownershipState.isPresent()
331 || Objects.equals(ownershipState.get(), EntityOwnershipState.NO_OWNER)) {
332 LOG.debug("Entity {} has no owner", entityName);
333 final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
334 DeviceStateUtil.createNodeInstanceIdentifier(new NodeId(entityName));
335 deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
336 LOG.info("Try to remove device {} from operational DS", entityName);
337 ListenableFuture<?> future =
338 deviceManager.removeDeviceFromOperationalDS(nodeInstanceIdentifier);
339 Futures.addCallback(future, new FutureCallback<Object>() {
341 public void onSuccess(final Object result) {
342 LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
343 OF_EVENT_LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
347 public void onFailure(final Throwable throwable) {
348 LOG.error("Could not remove device {} from operational DS", dpnId, throwable);
350 }, MoreExecutors.directExecutor());
351 future.get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
353 LOG.warn("Seems like device is still owned by other controller instance. Skip deleting {} "
354 + "node from operational datastore.", entityName);
356 } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
357 LOG.warn("Not able to remove device {} from operational DS. ", entityName, e);
359 }, config.getDeviceDatastoreRemovalDelay().getValue().toJava(), TimeUnit.MILLISECONDS);
363 private void destroyContextChain(final DeviceInfo deviceInfo) {
364 OF_EVENT_LOG.debug("Destroying context chain for device {}", deviceInfo.getDatapathId());
365 ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
366 Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
367 deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
368 contextChain.close();
369 connectingDevices.remove(deviceInfo);
374 public List<DeviceInfo> listOfMasteredDevices() {
375 return contextChainMap.entrySet().stream()
376 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
377 .isMastered(ContextChainMastershipState.CHECK, false)).map(Map.Entry::getKey)
378 .collect(Collectors.toList());
382 public boolean isAnyDeviceMastered() {
383 return contextChainMap.entrySet().stream().findAny()
384 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
385 .isMastered(ContextChainMastershipState.CHECK, false)).isPresent();
389 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
390 contextChainMap.remove(deviceInfo);
391 LOG.debug("Context chain removed for node {}", deviceInfo);
394 private FutureCallback<ResultState> reconciliationFrameworkCallback(@NonNull final DeviceInfo deviceInfo,
395 final ContextChain contextChain, final ContextChainMastershipState mastershipState) {
396 return new FutureCallback<>() {
398 public void onSuccess(final ResultState result) {
399 if (ResultState.DONOTHING == result) {
400 OF_EVENT_LOG.debug("Device {} connection is enabled by reconciliation framework", deviceInfo);
401 LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
402 if (mastershipState == ContextChainMastershipState.MASTER_ON_DEVICE) {
403 ownershipChangeListener.becomeMaster(deviceInfo);
404 contextChain.initializeDevice();
406 contextChain.continueInitializationAfterReconciliation();
408 OF_EVENT_LOG.debug("Reconciliation framework failure for device {}", deviceInfo);
409 LOG.warn("Reconciliation framework failure for device {} with resultState {}", deviceInfo, result);
410 destroyContextChain(deviceInfo);
415 public void onFailure(final Throwable throwable) {
416 OF_EVENT_LOG.debug("Reconciliation framework failure for device {} with error {}", deviceInfo,
417 throwable.getMessage());
418 LOG.warn("Reconciliation framework failure for device {}", deviceInfo, throwable);
419 destroyContextChain(deviceInfo);
424 private String getDpnIdFromNodeName(final String nodeName) {
425 return nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1);
428 private Optional<EntityOwnershipState> getCurrentOwnershipStatus(final String nodeId) {
429 org.opendaylight.mdsal.eos.binding.api.Entity entity = createNodeEntity(nodeId);
430 Optional<EntityOwnershipState> ownershipStatus
431 = entityOwnershipService.getOwnershipState(entity);
433 if (ownershipStatus.isPresent()) {
434 LOG.debug("Current ownership status for node {} is {}", nodeId, ownershipStatus.get());
435 return Optional.of(ownershipStatus.get());
437 LOG.trace("Ownership status is not available for node {}", nodeId);
439 return Optional.empty();
442 private org.opendaylight.mdsal.eos.binding.api.Entity createNodeEntity(final String nodeId) {
443 return new org.opendaylight.mdsal.eos.binding.api.Entity(ASYNC_SERVICE_ENTITY_TYPE, nodeId);