openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/role/RoleContextImpl.java

   1 /*
   2  * Copyright (c) 2017 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
   9 package org.opendaylight.openflowplugin.impl.role;
  10
  11 import com.google.common.util.concurrent.FutureCallback;
  12 import com.google.common.util.concurrent.Futures;
  13 import com.google.common.util.concurrent.JdkFutureAdapters;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import io.netty.util.HashedWheelTimer;
  17 import io.netty.util.Timeout;
  18 import io.netty.util.TimerTask;
  19 import java.util.Collection;
  20 import java.util.HashSet;
  21 import java.util.Objects;
  22 import java.util.concurrent.CancellationException;
  23 import java.util.concurrent.Future;
  24 import java.util.concurrent.TimeUnit;
  25 import java.util.concurrent.atomic.AtomicReference;
  26 import javax.annotation.Nonnull;
  27 import javax.annotation.Nullable;
  28 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  29 import org.opendaylight.openflowplugin.api.OFConstants;
  30 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  31 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  32 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
  33 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipWatcher;
  34 import org.opendaylight.openflowplugin.api.openflow.role.RoleContext;
  35 import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
  36 import org.opendaylight.openflowplugin.impl.services.util.RequestContextUtil;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SalRoleService;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInput;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInputBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleOutput;
  43 import org.opendaylight.yangtools.yang.common.RpcResult;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46
  47 public class RoleContextImpl implements RoleContext {
  48     private static final Logger LOG = LoggerFactory.getLogger(RoleContextImpl.class);
  49
  50     // Timeout  after what we will give up on propagating role
  51     private static final long SET_ROLE_TIMEOUT = 10000;
  52
  53     private final DeviceInfo deviceInfo;
  54     private final HashedWheelTimer timer;
  55     private final AtomicReference<ListenableFuture<RpcResult<SetRoleOutput>>> lastRoleFuture = new AtomicReference<>();
  56     private final Collection<RequestContext<?>> requestContexts = new HashSet<>();
  57     private final Timeout slaveTask;
  58     private ContextChainMastershipWatcher contextChainMastershipWatcher;
  59     private SalRoleService roleService;
  60
  61     RoleContextImpl(@Nonnull final DeviceInfo deviceInfo,
  62                     @Nonnull final HashedWheelTimer timer,
  63                     final long checkRoleMasterTimeout) {
  64         this.deviceInfo = deviceInfo;
  65         this.timer = timer;
  66         slaveTask = timer.newTimeout((timerTask) -> makeDeviceSlave(), checkRoleMasterTimeout, TimeUnit.MILLISECONDS);
  67
  68         LOG.info("Started timer for setting SLAVE role on device {} if no role will be set in {}s.",
  69                 deviceInfo,
  70                 checkRoleMasterTimeout / 1000L);
  71     }
  72
  73     @Override
  74     public DeviceInfo getDeviceInfo() {
  75         return deviceInfo;
  76     }
  77
  78     @Override
  79     public void setRoleService(final SalRoleService salRoleService) {
  80         roleService = salRoleService;
  81     }
  82
  83     @Override
  84     public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher newWatcher) {
  85         this.contextChainMastershipWatcher = newWatcher;
  86     }
  87
  88     @Override
  89     public void close() {
  90         changeLastRoleFuture(null);
  91         requestContexts.forEach(requestContext -> RequestContextUtil
  92                 .closeRequestContextWithRpcError(requestContext, "Connection closed."));
  93         requestContexts.clear();
  94     }
  95
  96     @Override
  97     public void instantiateServiceInstance() {
  98         final ListenableFuture<RpcResult<SetRoleOutput>> future = sendRoleChangeToDevice(OfpRole.BECOMEMASTER);
  99         changeLastRoleFuture(future);
 100         Futures.addCallback(future, new MasterRoleCallback(), MoreExecutors.directExecutor());
 101     }
 102
 103     @Override
 104     public ListenableFuture<Void> closeServiceInstance() {
 105         changeLastRoleFuture(null);
 106         return Futures.immediateFuture(null);
 107     }
 108
 109     @Override
 110     public <T> RequestContext<T> createRequestContext() {
 111         final AbstractRequestContext<T> ret = new AbstractRequestContext<T>(deviceInfo.reserveXidForDeviceMessage()) {
 112             @Override
 113             public void close() {
 114                 requestContexts.remove(this);
 115             }
 116         };
 117
 118         requestContexts.add(ret);
 119         return ret;
 120     }
 121
 122     @Nonnull
 123     @Override
 124     public ServiceGroupIdentifier getIdentifier() {
 125         return deviceInfo.getServiceIdentifier();
 126     }
 127
 128     private void changeLastRoleFuture(final ListenableFuture<RpcResult<SetRoleOutput>> newFuture) {
 129         slaveTask.cancel();
 130         lastRoleFuture.getAndUpdate(lastFuture -> {
 131             if (Objects.nonNull(lastFuture) && !lastFuture.isCancelled() && !lastFuture.isDone()) {
 132                 lastFuture.cancel(true);
 133             }
 134
 135             return newFuture;
 136         });
 137     }
 138
 139     private ListenableFuture<RpcResult<SetRoleOutput>> makeDeviceSlave() {
 140         final ListenableFuture<RpcResult<SetRoleOutput>> future = sendRoleChangeToDevice(OfpRole.BECOMESLAVE);
 141         changeLastRoleFuture(future);
 142         Futures.addCallback(future, new SlaveRoleCallback(), MoreExecutors.directExecutor());
 143         return future;
 144     }
 145
 146     private ListenableFuture<RpcResult<SetRoleOutput>> sendRoleChangeToDevice(final OfpRole newRole) {
 147         LOG.debug("Sending new role {} to device {}", newRole, deviceInfo);
 148
 149         if (deviceInfo.getVersion() >= OFConstants.OFP_VERSION_1_3) {
 150             final SetRoleInput setRoleInput = new SetRoleInputBuilder()
 151                     .setControllerRole(newRole)
 152                     .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier()))
 153                     .build();
 154
 155             final Future<RpcResult<SetRoleOutput>> setRoleOutputFuture = roleService.setRole(setRoleInput);
 156
 157             final TimerTask timerTask = timeout -> {
 158                 if (!setRoleOutputFuture.isDone()) {
 159                     LOG.warn("New role {} was not propagated to device {} during {} sec", newRole,
 160                             deviceInfo, SET_ROLE_TIMEOUT);
 161                     setRoleOutputFuture.cancel(true);
 162                 }
 163             };
 164
 165             timer.newTimeout(timerTask, SET_ROLE_TIMEOUT, TimeUnit.MILLISECONDS);
 166             return JdkFutureAdapters.listenInPoolThread(setRoleOutputFuture);
 167         }
 168
 169         LOG.info("Device: {} with version: {} does not support role {}", deviceInfo, deviceInfo.getVersion(), newRole);
 170         return Futures.immediateFuture(null);
 171     }
 172
 173     private final class MasterRoleCallback implements FutureCallback<RpcResult<SetRoleOutput>> {
 174         @Override
 175         public void onSuccess(@Nullable RpcResult<SetRoleOutput> setRoleOutputRpcResult) {
 176             contextChainMastershipWatcher.onMasterRoleAcquired(
 177                     deviceInfo,
 178                     ContextChainMastershipState.MASTER_ON_DEVICE);
 179             LOG.debug("Role MASTER was successfully set on device, node {}", deviceInfo);
 180         }
 181
 182         @Override
 183         public void onFailure(@Nonnull final Throwable throwable) {
 184             if (!(throwable instanceof CancellationException)) {
 185                 contextChainMastershipWatcher.onNotAbleToStartMastershipMandatory(
 186                         deviceInfo,
 187                         "Was not able to propagate MASTER role on device. Error: " + throwable.toString());
 188             }
 189         }
 190     }
 191
 192     private final class SlaveRoleCallback implements FutureCallback<RpcResult<SetRoleOutput>> {
 193         @Override
 194         public void onSuccess(@Nullable final RpcResult<SetRoleOutput> result) {
 195             contextChainMastershipWatcher.onSlaveRoleAcquired(deviceInfo);
 196             LOG.debug("Role SLAVE was successfully set on device, node {}", deviceInfo);
 197         }
 198
 199         @Override
 200         public void onFailure(@Nonnull final Throwable throwable) {
 201             if (!(throwable instanceof CancellationException)) {
 202                 contextChainMastershipWatcher.onSlaveRoleNotAcquired(deviceInfo,
 203                         "Was not able to propagate SLAVE role on device. Error: " + throwable.toString());
 204             }
 205         }
 206     }
 207 }