renderers/vpp/src/main/java/org/opendaylight/groupbasedpolicy/renderer/vpp/util/MountedDataBrokerProvider.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. 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
   9 package org.opendaylight.groupbasedpolicy.renderer.vpp.util;
  10
  11 import java.util.AbstractMap;
  12 import java.util.concurrent.Callable;
  13 import java.util.concurrent.ExecutionException;
  14 import java.util.concurrent.ExecutorService;
  15 import java.util.concurrent.Executors;
  16 import java.util.concurrent.Future;
  17 import java.util.concurrent.TimeUnit;
  18 import java.util.concurrent.TimeoutException;
  19 import java.util.concurrent.locks.ReentrantLock;
  20
  21 import javax.annotation.Nonnull;
  22
  23 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  24 import org.opendaylight.controller.md.sal.binding.api.MountPoint;
  25 import org.opendaylight.controller.md.sal.binding.api.MountPointService;
  26 import org.opendaylight.vbd.impl.transaction.VbdNetconfTransaction;
  27 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  28 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  29 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 import com.google.common.base.Optional;
  34 import com.google.common.base.Preconditions;
  35 import com.google.common.util.concurrent.SettableFuture;
  36
  37 public class MountedDataBrokerProvider {
  38
  39     private static final Logger LOG = LoggerFactory.getLogger(MountedDataBrokerProvider.class);
  40     private static final short DURATION = 3000;
  41     private final MountPointService mountService;
  42     private final DataBroker dataBroker;
  43     @SuppressWarnings("FieldCanBeLocal")
  44     private final byte NODE_CONNECTION_TIMER = 60; // seconds
  45
  46     public MountedDataBrokerProvider(@Nonnull MountPointService mountService, @Nonnull DataBroker dataBroker) {
  47         this.mountService = Preconditions.checkNotNull(mountService);
  48         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  49     }
  50
  51     public Optional<DataBroker> resolveDataBrokerForMountPoint(@Nonnull InstanceIdentifier<Node> iidToMountPoint) {
  52         try {
  53             if (VbdNetconfTransaction.NODE_DATA_BROKER_MAP.get(iidToMountPoint) != null) {
  54                 return Optional.of(VbdNetconfTransaction.NODE_DATA_BROKER_MAP.get(iidToMountPoint).getKey());
  55             }
  56             final SettableFuture<Boolean> futureNodeStatus = SettableFuture.create();
  57             final NodeKey nodeKey = iidToMountPoint.firstKeyOf(Node.class);
  58             new GbpVppNetconfConnectionProbe(nodeKey, futureNodeStatus, dataBroker);
  59             if (futureNodeStatus.get(NODE_CONNECTION_TIMER, TimeUnit.SECONDS)) {
  60                 LOG.debug("Node connected, mountpoint with iid {} available", iidToMountPoint);
  61                 Future<Optional<MountPoint>> mountPointfuture = getMountpointFromSal(iidToMountPoint);
  62                 Optional<MountPoint> potentialMountPoint = mountPointfuture.get();
  63                 if (potentialMountPoint.isPresent()) {
  64                     final Optional<DataBroker> dataBrokerOpt = potentialMountPoint.get().getService(DataBroker.class);
  65                     VbdNetconfTransaction.NODE_DATA_BROKER_MAP.put(iidToMountPoint,
  66                             new AbstractMap.SimpleEntry(dataBrokerOpt.get(), new ReentrantLock()));
  67                     LOG.info("Lock created for {}", iidToMountPoint);
  68                     return dataBrokerOpt;
  69                 } else {
  70                     LOG.warn("Mount point does not exist for {}", iidToMountPoint);
  71                     return Optional.absent();
  72                 }
  73             } else {
  74                 LOG.warn("Failed while connecting to node, Iid: {}", iidToMountPoint);
  75                 return Optional.absent();
  76             }
  77         } catch (TimeoutException e) {
  78             LOG.warn("Mountpoint not obtained within {} seconds. Iid: {}", NODE_CONNECTION_TIMER, iidToMountPoint, e);
  79             return Optional.absent();
  80         } catch (ExecutionException | InterruptedException e) {
  81             LOG.warn("Error while getting mountpoint. Iid: {}", iidToMountPoint, e);
  82             return Optional.absent();
  83         }
  84     }
  85
  86     // TODO bug 7699
  87     // This works as a workaround for mountpoint registration in cluster. If application is registered on different
  88     // node as netconf service, it obtains mountpoint registered by SlaveSalFacade (instead of MasterSalFacade). However
  89     // this service registers mountpoint a moment later then connectionStatus is set to "Connected". If NodeManager hits
  90     // state where device is connected but mountpoint is not yet available, try to get it again in a while
  91     private Future<Optional<MountPoint>> getMountpointFromSal(final InstanceIdentifier<Node> iid) {
  92         final ExecutorService executorService = Executors.newSingleThreadExecutor();
  93         final Callable<Optional<MountPoint>> task = () -> {
  94             byte attempt = 0;
  95             do {
  96                 try {
  97                     final Optional<MountPoint> optionalMountpoint = mountService.getMountPoint(iid);
  98                     if (optionalMountpoint.isPresent()) {
  99                         return optionalMountpoint;
 100                     }
 101                     LOG.warn("Mountpoint {} is not registered yet", iid);
 102                     Thread.sleep(DURATION);
 103                 } catch (InterruptedException e) {
 104                     LOG.warn("Thread interrupted to ", e);
 105                 }
 106                 attempt++;
 107             } while (attempt <= 3);
 108             return Optional.absent();
 109         };
 110         return executorService.submit(task);
 111     }
 112 }