73bb3fe0c65294220f07549411ba8972f2ad70c1
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / lifecycle / ContextChainHolderImpl.java
1 /*
2  * Copyright (c) 2016 Pantheon Technologies s.r.o. 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 package org.opendaylight.openflowplugin.impl.lifecycle;
9
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;
19 import java.util.Map;
20 import java.util.Objects;
21 import java.util.Optional;
22 import java.util.concurrent.ConcurrentHashMap;
23 import java.util.concurrent.ConcurrentMap;
24 import java.util.concurrent.ExecutionException;
25 import java.util.concurrent.Executor;
26 import java.util.concurrent.Executors;
27 import java.util.concurrent.ScheduledExecutorService;
28 import java.util.concurrent.ThreadFactory;
29 import java.util.concurrent.TimeUnit;
30 import java.util.concurrent.TimeoutException;
31 import java.util.stream.Collectors;
32 import org.eclipse.jdt.annotation.NonNull;
33 import org.eclipse.jdt.annotation.Nullable;
34 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipChange;
35 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipListenerRegistration;
36 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipService;
37 import org.opendaylight.mdsal.eos.common.api.EntityOwnershipState;
38 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
39 import org.opendaylight.openflowplugin.api.openflow.OFPManager;
40 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
41 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionStatus;
42 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
43 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
44 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
45 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
46 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
47 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
48 import org.opendaylight.openflowplugin.api.openflow.lifecycle.MasterChecker;
49 import org.opendaylight.openflowplugin.api.openflow.lifecycle.OwnershipChangeListener;
50 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
51 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
52 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
53 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
54 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
55 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
56 import org.opendaylight.openflowplugin.impl.util.DeviceStateUtil;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.mdsal.core.general.entity.rev150930.Entity;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
64 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
65 import org.opendaylight.yangtools.yang.common.Uint8;
66 import org.slf4j.Logger;
67 import org.slf4j.LoggerFactory;
68
69 public class ContextChainHolderImpl implements ContextChainHolder, MasterChecker {
70     private static final Logger LOG = LoggerFactory.getLogger(ContextChainHolderImpl.class);
71     private static final Logger OF_EVENT_LOG = LoggerFactory.getLogger("OfEventLog");
72
73     private static final String CONTEXT_CREATED_FOR_CONNECTION = " context created for connection: {}";
74     private static final long REMOVE_DEVICE_FROM_DS_TIMEOUT = 5000L;
75     private static final String ASYNC_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
76     private static final String SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.ServiceEntityType";
77     private static final String SEPARATOR = ":";
78     private final ConcurrentMap<DeviceInfo, ContextChain> contextChainMap = new ConcurrentHashMap<>();
79     private final ConcurrentMap<DeviceInfo, ? super ConnectionContext> connectingDevices = new ConcurrentHashMap<>();
80     private final EntityOwnershipListenerRegistration eosListenerRegistration;
81     private final ClusterSingletonServiceProvider singletonServiceProvider;
82     private final Executor executor;
83     private final OwnershipChangeListener ownershipChangeListener;
84     private final ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true)
85             .setNameFormat("node-cleaner-%d").setUncaughtExceptionHandler((thread, throwable) -> {
86                 LOG.warn("Uncaught exception while removing node data from operational datastore.", throwable);
87             }).build();
88     private final ScheduledExecutorService nodeCleanerExecutor = Executors.newScheduledThreadPool(
89             Runtime.getRuntime().availableProcessors() , threadFactory);
90     private final EntityOwnershipService entityOwnershipService;
91     private final OpenflowProviderConfig config;
92     private DeviceManager deviceManager;
93     private RpcManager rpcManager;
94     private StatisticsManager statisticsManager;
95     private RoleManager roleManager;
96
97     public ContextChainHolderImpl(final Executor executor,
98                                   final ClusterSingletonServiceProvider singletonServiceProvider,
99                                   final EntityOwnershipService entityOwnershipService,
100                                   final OwnershipChangeListener ownershipChangeListener,
101                                   final OpenflowProviderConfig config) {
102         this.singletonServiceProvider = singletonServiceProvider;
103         this.executor = executor;
104         this.ownershipChangeListener = ownershipChangeListener;
105         this.ownershipChangeListener.setMasterChecker(this);
106         this.entityOwnershipService = entityOwnershipService;
107         this.config = config;
108         this.eosListenerRegistration = Objects
109                 .requireNonNull(entityOwnershipService.registerListener(ASYNC_SERVICE_ENTITY_TYPE, this));
110     }
111
112     @Override
113     public <T extends OFPManager> void addManager(final T manager) {
114         if (deviceManager == null && manager instanceof DeviceManager) {
115             LOG.trace("Context chain holder: Device manager OK.");
116             deviceManager = (DeviceManager) manager;
117         } else if (rpcManager == null && manager instanceof RpcManager) {
118             LOG.trace("Context chain holder: RPC manager OK.");
119             rpcManager = (RpcManager) manager;
120         } else if (statisticsManager == null && manager instanceof StatisticsManager) {
121             LOG.trace("Context chain holder: Statistics manager OK.");
122             statisticsManager = (StatisticsManager) manager;
123         } else if (roleManager == null && manager instanceof RoleManager) {
124             LOG.trace("Context chain holder: Role manager OK.");
125             roleManager = (RoleManager) manager;
126         }
127     }
128
129     @VisibleForTesting
130     void createContextChain(final ConnectionContext connectionContext) {
131         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
132         final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
133         deviceContext.registerMastershipWatcher(this);
134         LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
135
136         final RpcContext rpcContext = rpcManager.createContext(deviceContext);
137         rpcContext.registerMastershipWatcher(this);
138         LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
139
140         final StatisticsContext statisticsContext = statisticsManager
141                 .createContext(deviceContext, ownershipChangeListener.isReconciliationFrameworkRegistered());
142         statisticsContext.registerMastershipWatcher(this);
143         LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
144
145         final RoleContext roleContext = roleManager.createContext(deviceContext);
146         roleContext.registerMastershipWatcher(this);
147         LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
148
149         final ContextChain contextChain = new ContextChainImpl(this, connectionContext, executor);
150         contextChain.registerDeviceRemovedHandler(deviceManager);
151         contextChain.registerDeviceRemovedHandler(rpcManager);
152         contextChain.registerDeviceRemovedHandler(statisticsManager);
153         contextChain.registerDeviceRemovedHandler(roleManager);
154         contextChain.registerDeviceRemovedHandler(this);
155         contextChain.addContext(deviceContext);
156         contextChain.addContext(rpcContext);
157         contextChain.addContext(statisticsContext);
158         contextChain.addContext(roleContext);
159         contextChainMap.put(deviceInfo, contextChain);
160         connectingDevices.remove(deviceInfo);
161         LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
162
163         deviceContext.onPublished();
164         contextChain.registerServices(singletonServiceProvider);
165     }
166
167     @Override
168     public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) {
169         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
170         final ContextChain contextChain = contextChainMap.get(deviceInfo);
171         final FeaturesReply featuresReply = connectionContext.getFeatures();
172         final Uint8 auxiliaryId = featuresReply != null ? featuresReply.getAuxiliaryId() : null;
173
174         if (auxiliaryId != null && auxiliaryId.toJava() != 0) {
175             if (contextChain == null) {
176                 LOG.warn("An auxiliary connection for device {}, but no primary connection. Refusing connection.",
177                          deviceInfo);
178                 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
179             } else {
180                 if (contextChain.addAuxiliaryConnection(connectionContext)) {
181                     LOG.info("An auxiliary connection was added to device: {}", deviceInfo);
182                     return ConnectionStatus.MAY_CONTINUE;
183                 } else {
184                     LOG.warn("Not able to add auxiliary connection to the device {}", deviceInfo);
185                     return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
186                 }
187             }
188         } else {
189             LOG.info("Device {} connected.", deviceInfo);
190             final boolean contextExists = contextChain != null;
191             final boolean isClosing = contextExists && contextChain.isClosing();
192
193             if (!isClosing && connectingDevices.putIfAbsent(deviceInfo, connectionContext) != null) {
194                 LOG.warn("Device {} is already trying to connect, wait until succeeded or disconnected.", deviceInfo);
195                 return ConnectionStatus.ALREADY_CONNECTED;
196             }
197
198             if (contextExists) {
199                 if (isClosing) {
200                     LOG.warn("Device {} is already in termination state, closing all incoming connections.",
201                              deviceInfo);
202                     return ConnectionStatus.CLOSING;
203                 }
204
205                 LOG.warn("Device {} already connected. Closing previous connection", deviceInfo);
206                 destroyContextChain(deviceInfo);
207                 LOG.info("Old connection dropped, creating new context chain for device {}", deviceInfo);
208                 createContextChain(connectionContext);
209             } else {
210                 LOG.info("No context chain found for device: {}, creating new.", deviceInfo);
211                 createContextChain(connectionContext);
212             }
213
214             return ConnectionStatus.MAY_CONTINUE;
215         }
216
217     }
218
219     @Override
220     public void onNotAbleToStartMastership(final DeviceInfo deviceInfo, final String reason, final boolean mandatory) {
221         LOG.error("Not able to set MASTER role on device {}, reason: {}", deviceInfo, reason);
222
223         if (!mandatory) {
224             return;
225         }
226         if (contextChainMap.containsKey(deviceInfo)) {
227             LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.",
228                      deviceInfo);
229             destroyContextChain(deviceInfo);
230         }
231     }
232
233     @Override
234     public void onMasterRoleAcquired(final DeviceInfo deviceInfo, final ContextChainMastershipState mastershipState) {
235         final ContextChain contextChain = contextChainMap.get(deviceInfo);
236         if (contextChain != null) {
237             if (!ContextChainMastershipState.INITIAL_SUBMIT.equals(mastershipState)) {
238                 if (contextChain.isMastered(mastershipState, true)) {
239                     Futures.addCallback(ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
240                                         reconciliationFrameworkCallback(deviceInfo, contextChain, mastershipState),
241                                         MoreExecutors.directExecutor());
242                 }
243             } else if (contextChain.isMastered(mastershipState, false)) {
244                 LOG.info("Role MASTER was granted to device {}", deviceInfo);
245                 OF_EVENT_LOG.debug("Master Elected, Node: {}", deviceInfo.getDatapathId());
246                 deviceManager.sendNodeAddedNotification(deviceInfo.getNodeInstanceIdentifier());
247             }
248         }
249     }
250
251     @Override
252     public void onSlaveRoleAcquired(final DeviceInfo deviceInfo) {
253         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
254         LOG.info("Role SLAVE was granted to device {}", deviceInfo);
255         final ContextChain contextChain = contextChainMap.get(deviceInfo);
256         if (contextChain != null) {
257             contextChain.makeContextChainStateSlave();
258         }
259     }
260
261     @Override
262     public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
263         LOG.error("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
264         if (contextChainMap.containsKey(deviceInfo)) {
265             destroyContextChain(deviceInfo);
266         }
267     }
268
269     @Override
270     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
271         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
272         if (deviceInfo != null) {
273             final ContextChain contextChain = contextChainMap.get(deviceInfo);
274             if (contextChain != null) {
275                 if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
276                     LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
277                 } else {
278                     LOG.info("Device {} disconnected.", deviceInfo);
279                     destroyContextChain(deviceInfo);
280                 }
281             }
282         }
283     }
284
285     @VisibleForTesting
286     boolean checkAllManagers() {
287         return deviceManager != null && rpcManager != null && statisticsManager != null && roleManager != null;
288     }
289
290     @Override
291     public ContextChain getContextChain(final DeviceInfo deviceInfo) {
292         return contextChainMap.get(deviceInfo);
293     }
294
295     @Override
296     public void close() {
297         Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
298         copyOfChains.keySet().forEach(this::destroyContextChain);
299         copyOfChains.clear();
300         contextChainMap.clear();
301         OF_EVENT_LOG.debug("EOS registration closed for all devices");
302         eosListenerRegistration.close();
303         OF_EVENT_LOG.debug("EOS registration closed for all devices");
304         nodeCleanerExecutor.shutdownNow();
305     }
306
307     @Override
308     @SuppressFBWarnings("BC_UNCONFIRMED_CAST_OF_RETURN_VALUE")
309     public void ownershipChanged(final EntityOwnershipChange entityOwnershipChange) {
310         LOG.info("Entity ownership change received for node : {}", entityOwnershipChange);
311         if (entityOwnershipChange.inJeopardy()) {
312             LOG.warn("Controller is in Jeopardy, ignore ownership change notification. {}", entityOwnershipChange);
313             return;
314         }
315         if (entityOwnershipChange.getState().hasOwner()) {
316             return;
317         }
318
319         // Findbugs flags a false violation for "Unchecked/unconfirmed cast" from GenericEntity to Entity hence the
320         // suppression above. The suppression is temporary until EntityOwnershipChange is modified to eliminate the
321         // violation.
322         final String entityName = entityOwnershipChange
323                 .getEntity()
324                 .getIdentifier()
325                 .firstKeyOf(Entity.class)
326                 .getName();
327
328         if (entityName != null && entityName.startsWith("openflow:")) {
329             if (nodeCleanerExecutor.isShutdown()) {
330                 LOG.warn("Node cleaner executor thread-pool is down.");
331                 return;
332             }
333             LOG.debug("Device {} will be removed from datastore in {} msec, if it's not transient notification.",
334                     entityName, config.getDeviceDatastoreRemovalDelay().getValue());
335             final String dpnId = getDpnIdFromNodeName(entityName);
336             nodeCleanerExecutor.schedule(() -> {
337                 try {
338                     EntityOwnershipState ownershipState = getCurrentOwnershipStatus(entityName);
339                     if (ownershipState == null || EntityOwnershipState.NO_OWNER.equals(ownershipState)) {
340                         LOG.debug("Entity {} has no owner", entityName);
341                         final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
342                                 DeviceStateUtil.createNodeInstanceIdentifier(new NodeId(entityName));
343                         deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
344                         LOG.info("Try to remove device {} from operational DS", entityName);
345                         ListenableFuture<?> future =
346                                 deviceManager.removeDeviceFromOperationalDS(nodeInstanceIdentifier);
347                         Futures.addCallback(future, new FutureCallback<Object>() {
348                             @Override
349                             public void onSuccess(final Object result) {
350                                 LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
351                                 OF_EVENT_LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
352                             }
353
354                             @Override
355                             public void onFailure(final Throwable throwable) {
356                                 LOG.error("Could not remove device {} from operational DS", dpnId, throwable);
357                             }
358                         }, MoreExecutors.directExecutor());
359                         future.get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
360                     } else {
361                         LOG.warn("Seems like device is still owned by other controller instance. Skip deleting {} "
362                                 + "node from operational datastore.", entityName);
363                     }
364                 } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
365                     LOG.warn("Not able to remove device {} from operational DS. ", entityName, e);
366                 }
367             }, config.getDeviceDatastoreRemovalDelay().getValue().toJava(), TimeUnit.MILLISECONDS);
368         }
369     }
370
371     private void destroyContextChain(final DeviceInfo deviceInfo) {
372         OF_EVENT_LOG.debug("Destroying context chain for device {}", deviceInfo.getDatapathId());
373         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
374         final ContextChain contextChain = contextChainMap.get(deviceInfo);
375         if (contextChain != null) {
376             deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
377             contextChain.close();
378             connectingDevices.remove(deviceInfo);
379         }
380     }
381
382     @Override
383     public List<DeviceInfo> listOfMasteredDevices() {
384         return contextChainMap.entrySet().stream()
385                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
386                         .isMastered(ContextChainMastershipState.CHECK, false)).map(Map.Entry::getKey)
387                 .collect(Collectors.toList());
388     }
389
390     @Override
391     public boolean isAnyDeviceMastered() {
392         return contextChainMap.entrySet().stream().findAny()
393                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
394                         .isMastered(ContextChainMastershipState.CHECK, false)).isPresent();
395     }
396
397     @Override
398     public void onDeviceRemoved(final DeviceInfo deviceInfo) {
399         contextChainMap.remove(deviceInfo);
400         LOG.debug("Context chain removed for node {}", deviceInfo);
401     }
402
403     private FutureCallback<ResultState> reconciliationFrameworkCallback(@NonNull final DeviceInfo deviceInfo,
404             final ContextChain contextChain, final ContextChainMastershipState mastershipState) {
405         return new FutureCallback<>() {
406             @Override
407             public void onSuccess(final ResultState result) {
408                 if (ResultState.DONOTHING == result) {
409                     OF_EVENT_LOG.debug("Device {} connection is enabled by reconciliation framework", deviceInfo);
410                     LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
411                     if (mastershipState == ContextChainMastershipState.MASTER_ON_DEVICE) {
412                         ownershipChangeListener.becomeMaster(deviceInfo);
413                         contextChain.initializeDevice();
414                     }
415                     contextChain.continueInitializationAfterReconciliation();
416                 } else {
417                     OF_EVENT_LOG.debug("Reconciliation framework failure for device {}", deviceInfo);
418                     LOG.warn("Reconciliation framework failure for device {} with resultState {}", deviceInfo, result);
419                     destroyContextChain(deviceInfo);
420                 }
421             }
422
423             @Override
424             public void onFailure(final Throwable throwable) {
425                 OF_EVENT_LOG.debug("Reconciliation framework failure for device {} with error {}", deviceInfo,
426                         throwable.getMessage());
427                 LOG.warn("Reconciliation framework failure for device {}", deviceInfo, throwable);
428                 destroyContextChain(deviceInfo);
429             }
430         };
431     }
432
433     private static String getDpnIdFromNodeName(final String nodeName) {
434         return nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1);
435     }
436
437     private @Nullable EntityOwnershipState getCurrentOwnershipStatus(final String nodeId) {
438         org.opendaylight.mdsal.eos.binding.api.Entity entity = createNodeEntity(nodeId);
439         Optional<EntityOwnershipState> ownershipStatus
440                 = entityOwnershipService.getOwnershipState(entity);
441
442         if (ownershipStatus.isPresent()) {
443             LOG.debug("Current ownership status for node {} is {}", nodeId, ownershipStatus.get());
444             return ownershipStatus.get();
445         }
446
447         LOG.trace("Ownership status is not available for node {}", nodeId);
448         return null;
449     }
450
451     private static org.opendaylight.mdsal.eos.binding.api.Entity createNodeEntity(final String nodeId) {
452         return new org.opendaylight.mdsal.eos.binding.api.Entity(ASYNC_SERVICE_ENTITY_TYPE, nodeId);
453     }
454 }