4393b96537ecd575d075f3dd05702a1bf8c8d1fa
[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.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;
66
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");
70
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);
85             }).build();
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;
94
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));
108     }
109
110     @Override
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;
124         }
125     }
126
127     @VisibleForTesting
128     void createContextChain(final ConnectionContext connectionContext) {
129         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
130
131         final DeviceContext deviceContext = deviceManager.createContext(connectionContext);
132         deviceContext.registerMastershipWatcher(this);
133         LOG.debug("Device" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
134
135         final RpcContext rpcContext = rpcManager.createContext(deviceContext);
136         rpcContext.registerMastershipWatcher(this);
137         LOG.debug("RPC" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
138
139         final StatisticsContext statisticsContext = statisticsManager
140                 .createContext(deviceContext, ownershipChangeListener.isReconciliationFrameworkRegistered());
141         statisticsContext.registerMastershipWatcher(this);
142         LOG.debug("Statistics" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
143
144         final RoleContext roleContext = roleManager.createContext(deviceContext);
145         roleContext.registerMastershipWatcher(this);
146         LOG.debug("Role" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
147
148         final ContextChain contextChain = new ContextChainImpl(this, connectionContext, executorService);
149         contextChain.registerDeviceRemovedHandler(deviceManager);
150         contextChain.registerDeviceRemovedHandler(rpcManager);
151         contextChain.registerDeviceRemovedHandler(statisticsManager);
152         contextChain.registerDeviceRemovedHandler(roleManager);
153         contextChain.registerDeviceRemovedHandler(this);
154         contextChain.addContext(deviceContext);
155         contextChain.addContext(rpcContext);
156         contextChain.addContext(statisticsContext);
157         contextChain.addContext(roleContext);
158         contextChainMap.put(deviceInfo, contextChain);
159         connectingDevices.remove(deviceInfo);
160         LOG.debug("Context chain" + CONTEXT_CREATED_FOR_CONNECTION, deviceInfo);
161
162         deviceContext.onPublished();
163         contextChain.registerServices(singletonServiceProvider);
164     }
165
166     @Override
167     public ConnectionStatus deviceConnected(final ConnectionContext connectionContext) {
168         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
169         final ContextChain contextChain = contextChainMap.get(deviceInfo);
170         final FeaturesReply featuresReply = connectionContext.getFeatures();
171         final Uint8 auxiliaryId = featuresReply != null ? featuresReply.getAuxiliaryId() : null;
172
173         if (auxiliaryId != null && auxiliaryId.toJava() != 0) {
174             if (contextChain == null) {
175                 LOG.warn("An auxiliary connection for device {}, but no primary connection. Refusing connection.",
176                          deviceInfo);
177                 return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
178             } else {
179                 if (contextChain.addAuxiliaryConnection(connectionContext)) {
180                     LOG.info("An auxiliary connection was added to device: {}", deviceInfo);
181                     return ConnectionStatus.MAY_CONTINUE;
182                 } else {
183                     LOG.warn("Not able to add auxiliary connection to the device {}", deviceInfo);
184                     return ConnectionStatus.REFUSING_AUXILIARY_CONNECTION;
185                 }
186             }
187         } else {
188             LOG.info("Device {} connected.", deviceInfo);
189             final boolean contextExists = contextChain != null;
190             final boolean isClosing = contextExists && contextChain.isClosing();
191
192             if (!isClosing && connectingDevices.putIfAbsent(deviceInfo, connectionContext) != null) {
193                 LOG.warn("Device {} is already trying to connect, wait until succeeded or disconnected.", deviceInfo);
194                 return ConnectionStatus.ALREADY_CONNECTED;
195             }
196
197             if (contextExists) {
198                 if (isClosing) {
199                     LOG.warn("Device {} is already in termination state, closing all incoming connections.",
200                              deviceInfo);
201                     return ConnectionStatus.CLOSING;
202                 }
203
204                 LOG.warn("Device {} already connected. Closing previous connection", deviceInfo);
205                 destroyContextChain(deviceInfo);
206                 LOG.info("Old connection dropped, creating new context chain for device {}", deviceInfo);
207                 createContextChain(connectionContext);
208             } else {
209                 LOG.info("No context chain found for device: {}, creating new.", deviceInfo);
210                 createContextChain(connectionContext);
211             }
212
213             return ConnectionStatus.MAY_CONTINUE;
214         }
215
216     }
217
218     @Override
219     public void onNotAbleToStartMastership(@NonNull final DeviceInfo deviceInfo, @NonNull final String reason,
220                                            final boolean mandatory) {
221         LOG.warn("Not able to set MASTER role on device {}, reason: {}", deviceInfo, reason);
222
223         if (!mandatory) {
224             return;
225         }
226
227         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
228             LOG.warn("This mastering is mandatory, destroying context chain and closing connection for device {}.",
229                      deviceInfo);
230             destroyContextChain(deviceInfo);
231         });
232     }
233
234     @Override
235     public void onMasterRoleAcquired(@NonNull final DeviceInfo deviceInfo,
236                                      @NonNull final ContextChainMastershipState mastershipState) {
237         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
238             if (!ContextChainMastershipState.INITIAL_SUBMIT.equals(mastershipState)) {
239                 if (contextChain.isMastered(mastershipState, true)) {
240                     Futures.addCallback(ownershipChangeListener.becomeMasterBeforeSubmittedDS(deviceInfo),
241                                         reconciliationFrameworkCallback(deviceInfo, contextChain, mastershipState),
242                                         MoreExecutors.directExecutor());
243                 }
244             } else if (contextChain.isMastered(mastershipState, false)) {
245                 LOG.info("Role MASTER was granted to device {}", deviceInfo);
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         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(ContextChain::makeContextChainStateSlave);
256     }
257
258     @Override
259     public void onSlaveRoleNotAcquired(final DeviceInfo deviceInfo, final String reason) {
260         LOG.warn("Not able to set SLAVE role on device {}, reason: {}", deviceInfo, reason);
261         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> destroyContextChain(deviceInfo));
262     }
263
264     @Override
265     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
266         final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
267
268         Optional.ofNullable(connectionContext.getDeviceInfo()).map(contextChainMap::get).ifPresent(contextChain -> {
269             if (contextChain.auxiliaryConnectionDropped(connectionContext)) {
270                 LOG.info("Auxiliary connection from device {} disconnected.", deviceInfo);
271             } else {
272                 LOG.info("Device {} disconnected.", deviceInfo);
273                 destroyContextChain(deviceInfo);
274             }
275         });
276     }
277
278     @VisibleForTesting
279     boolean checkAllManagers() {
280         return deviceManager != null && rpcManager != null && statisticsManager != null && roleManager != null;
281     }
282
283     @Override
284     public ContextChain getContextChain(final DeviceInfo deviceInfo) {
285         return contextChainMap.get(deviceInfo);
286     }
287
288     @Override
289     public void close() {
290         Map<DeviceInfo, ContextChain> copyOfChains = new HashMap<>(contextChainMap);
291         copyOfChains.keySet().forEach(this::destroyContextChain);
292         copyOfChains.clear();
293         contextChainMap.clear();
294         eosListenerRegistration.close();
295         OF_EVENT_LOG.debug("EOS registration closed for all devices");
296         nodeCleanerExecutor.shutdownNow();
297     }
298
299     @Override
300     @SuppressFBWarnings("BC_UNCONFIRMED_CAST_OF_RETURN_VALUE")
301     public void ownershipChanged(final EntityOwnershipChange entityOwnershipChange) {
302         LOG.info("Entity ownership change received for node : {}", entityOwnershipChange);
303         if (entityOwnershipChange.inJeopardy()) {
304             LOG.warn("Controller is in Jeopardy, ignore ownership change notification. {}", entityOwnershipChange);
305             return;
306         }
307         if (entityOwnershipChange.getState().hasOwner()) {
308             return;
309         }
310
311         // Findbugs flags a false violation for "Unchecked/unconfirmed cast" from GenericEntity to Entity hence the
312         // suppression above. The suppression is temporary until EntityOwnershipChange is modified to eliminate the
313         // violation.
314         final String entityName = entityOwnershipChange
315                 .getEntity()
316                 .getIdentifier()
317                 .firstKeyOf(Entity.class)
318                 .getName();
319
320         if (entityName != null && entityName.startsWith("openflow:")) {
321             if (nodeCleanerExecutor.isShutdown()) {
322                 LOG.warn("Node cleaner executor thread-pool is down.");
323                 return;
324             }
325             LOG.debug("Device {} will be removed from datastore in {} msec, if it's not transient notification.",
326                     entityName, config.getDeviceDatastoreRemovalDelay().getValue());
327             final String dpnId = getDpnIdFromNodeName(entityName);
328             nodeCleanerExecutor.schedule(() -> {
329                 try {
330                     Optional<EntityOwnershipState> ownershipState = getCurrentOwnershipStatus(entityName);
331                     if (!ownershipState.isPresent()
332                             || Objects.equals(ownershipState.get(), EntityOwnershipState.NO_OWNER)) {
333                         LOG.debug("Entity {} has no owner", entityName);
334                         final KeyedInstanceIdentifier<Node, NodeKey> nodeInstanceIdentifier =
335                                 DeviceStateUtil.createNodeInstanceIdentifier(new NodeId(entityName));
336                         deviceManager.sendNodeRemovedNotification(nodeInstanceIdentifier);
337                         LOG.info("Try to remove device {} from operational DS", entityName);
338                         ListenableFuture<?> future =
339                                 deviceManager.removeDeviceFromOperationalDS(nodeInstanceIdentifier);
340                         Futures.addCallback(future, new FutureCallback<Object>() {
341                             @Override
342                             public void onSuccess(final Object result) {
343                                 LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
344                                 OF_EVENT_LOG.debug("Node removed from Oper DS, Node: {}", dpnId);
345                             }
346
347                             @Override
348                             public void onFailure(final Throwable throwable) {
349                                 LOG.error("Could not remove device {} from operational DS", dpnId, throwable);
350                             }
351                         }, MoreExecutors.directExecutor());
352                         future.get(REMOVE_DEVICE_FROM_DS_TIMEOUT, TimeUnit.MILLISECONDS);
353                     } else {
354                         LOG.warn("Seems like device is still owned by other controller instance. Skip deleting {} "
355                                 + "node from operational datastore.", entityName);
356                     }
357                 } catch (TimeoutException | ExecutionException | NullPointerException | InterruptedException e) {
358                     LOG.warn("Not able to remove device {} from operational DS. ", entityName, e);
359                 }
360             }, config.getDeviceDatastoreRemovalDelay().getValue().toJava(), TimeUnit.MILLISECONDS);
361         }
362     }
363
364     private void destroyContextChain(final DeviceInfo deviceInfo) {
365         OF_EVENT_LOG.debug("Destroying context chain for device {}", deviceInfo.getDatapathId());
366         ownershipChangeListener.becomeSlaveOrDisconnect(deviceInfo);
367         Optional.ofNullable(contextChainMap.get(deviceInfo)).ifPresent(contextChain -> {
368             deviceManager.sendNodeRemovedNotification(deviceInfo.getNodeInstanceIdentifier());
369             contextChain.close();
370             connectingDevices.remove(deviceInfo);
371         });
372     }
373
374     @Override
375     public List<DeviceInfo> listOfMasteredDevices() {
376         return contextChainMap.entrySet().stream()
377                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
378                         .isMastered(ContextChainMastershipState.CHECK, false)).map(Map.Entry::getKey)
379                 .collect(Collectors.toList());
380     }
381
382     @Override
383     public boolean isAnyDeviceMastered() {
384         return contextChainMap.entrySet().stream().findAny()
385                 .filter(deviceInfoContextChainEntry -> deviceInfoContextChainEntry.getValue()
386                         .isMastered(ContextChainMastershipState.CHECK, false)).isPresent();
387     }
388
389     @Override
390     public void onDeviceRemoved(final DeviceInfo deviceInfo) {
391         contextChainMap.remove(deviceInfo);
392         LOG.debug("Context chain removed for node {}", deviceInfo);
393     }
394
395     private FutureCallback<ResultState> reconciliationFrameworkCallback(@NonNull final DeviceInfo deviceInfo,
396             final ContextChain contextChain, final ContextChainMastershipState mastershipState) {
397         return new FutureCallback<>() {
398             @Override
399             public void onSuccess(final ResultState result) {
400                 if (ResultState.DONOTHING == result) {
401                     OF_EVENT_LOG.debug("Device {} connection is enabled by reconciliation framework", deviceInfo);
402                     LOG.info("Device {} connection is enabled by reconciliation framework.", deviceInfo);
403                     if (mastershipState == ContextChainMastershipState.MASTER_ON_DEVICE) {
404                         ownershipChangeListener.becomeMaster(deviceInfo);
405                         contextChain.initializeDevice();
406                     }
407                     contextChain.continueInitializationAfterReconciliation();
408                 } else {
409                     OF_EVENT_LOG.debug("Reconciliation framework failure for device {}", deviceInfo);
410                     LOG.warn("Reconciliation framework failure for device {} with result {}", deviceInfo, result);
411                     destroyContextChain(deviceInfo);
412                 }
413             }
414
415             @Override
416             public void onFailure(final Throwable throwable) {
417                 OF_EVENT_LOG.debug("Reconciliation framework failure for device {} with error {}", deviceInfo,
418                         throwable.getMessage());
419                 LOG.warn("Reconciliation framework failure.", throwable);
420                 destroyContextChain(deviceInfo);
421             }
422         };
423     }
424
425     private String getDpnIdFromNodeName(final String nodeName) {
426         return nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1);
427     }
428
429     private Optional<EntityOwnershipState> getCurrentOwnershipStatus(final String nodeId) {
430         org.opendaylight.mdsal.eos.binding.api.Entity entity = createNodeEntity(nodeId);
431         Optional<EntityOwnershipState> ownershipStatus
432                 = entityOwnershipService.getOwnershipState(entity);
433
434         if (ownershipStatus.isPresent()) {
435             LOG.debug("Current ownership status for node {} is {}", nodeId, ownershipStatus.get());
436             return Optional.of(ownershipStatus.get());
437         } else {
438             LOG.trace("Ownership status is not available for node {}", nodeId);
439         }
440         return Optional.empty();
441     }
442
443     private org.opendaylight.mdsal.eos.binding.api.Entity createNodeEntity(final String nodeId) {
444         return new org.opendaylight.mdsal.eos.binding.api.Entity(ASYNC_SERVICE_ENTITY_TYPE, nodeId);
445     }
446 }