openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalRoleServiceImpl.java

   1 /**
   2  * Copyright (c) 2015 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 package org.opendaylight.openflowplugin.impl.services;
   9
  10 import com.google.common.base.Function;
  11 import com.google.common.util.concurrent.CheckedFuture;
  12 import com.google.common.util.concurrent.Futures;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.ListeningExecutorService;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import com.google.common.util.concurrent.SettableFuture;
  17 import java.math.BigInteger;
  18 import java.util.concurrent.Callable;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.Executors;
  21 import java.util.concurrent.Future;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import java.util.concurrent.atomic.AtomicReference;
  25 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  26 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  27 import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack;
  28 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  29 import org.opendaylight.openflowplugin.impl.role.RoleChangeException;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.RoleRequestOutput;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SalRoleService;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInput;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleOutput;
  37 import org.opendaylight.yangtools.yang.common.RpcResult;
  38 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42
  43 public class SalRoleServiceImpl extends AbstractSimpleService<SetRoleInput, SetRoleOutput> implements SalRoleService  {
  44
  45     private static final Logger LOG = LoggerFactory.getLogger(SalRoleServiceImpl.class);
  46
  47     private static final BigInteger MAX_GENERATION_ID = new BigInteger("ffffffffffffffff", 16);
  48
  49     private static final int MAX_RETRIES = 42;
  50
  51     private final DeviceContext deviceContext;
  52     private final RoleService roleService;
  53     private final AtomicReference<OfpRole> lastKnownRoleRef = new AtomicReference<>(OfpRole.NOCHANGE);
  54     private final ListeningExecutorService listeningExecutorService;
  55     private final NodeId nodeId;
  56     private final Short version;
  57
  58     public SalRoleServiceImpl(final RequestContextStack requestContextStack, final DeviceContext deviceContext) {
  59         super(requestContextStack, deviceContext, SetRoleOutput.class);
  60         this.deviceContext = deviceContext;
  61         this.roleService =  new RoleService(requestContextStack, deviceContext, RoleRequestOutput.class);
  62         nodeId = deviceContext.getPrimaryConnectionContext().getNodeId();
  63         version = deviceContext.getPrimaryConnectionContext().getFeatures().getVersion();
  64         listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
  65
  66     }
  67
  68     @Override
  69     protected OfHeader buildRequest(Xid xid, SetRoleInput input) {
  70         return null;
  71     }
  72
  73     public static BigInteger getNextGenerationId(BigInteger generationId) {
  74         BigInteger nextGenerationId = null;
  75         if (generationId.compareTo(MAX_GENERATION_ID) < 0) {
  76             nextGenerationId = generationId.add(BigInteger.ONE);
  77         } else {
  78             nextGenerationId = BigInteger.ZERO;
  79         }
  80
  81         return nextGenerationId;
  82     }
  83
  84
  85     @Override
  86     public Future<RpcResult<SetRoleOutput>> setRole(final SetRoleInput input) {
  87         LOG.info("SetRole called with input:{}", input);
  88         OfpRole lastKnownRole = lastKnownRoleRef.get();
  89
  90         // compare with last known role and set if different. If they are same, then return.
  91         if (lastKnownRoleRef.compareAndSet(input.getControllerRole(), input.getControllerRole())) {
  92             LOG.info("Role to be set is same as the last known role for the device:{}. Hence ignoring.", input.getControllerRole());
  93             SettableFuture<RpcResult<SetRoleOutput>> resultFuture = SettableFuture.create();
  94             resultFuture.set(RpcResultBuilder.<SetRoleOutput>success().build());
  95             return resultFuture;
  96         }
  97
  98         final SettableFuture<RpcResult<SetRoleOutput>> resultFuture = SettableFuture.create();
  99
 100         RoleChangeTask roleChangeTask = new RoleChangeTask(nodeId, input.getControllerRole(), version, roleService);
 101
 102         do {
 103             ListenableFuture<RpcResult<SetRoleOutput>> deviceCheck = deviceConnectionCheck();
 104             if (deviceCheck != null) {
 105                 LOG.info("Device {} is disconnected or state is not valid. Giving up on role change", input.getNode());
 106                 return deviceCheck;
 107             }
 108
 109             ListenableFuture<SetRoleOutput> taskFuture = listeningExecutorService.submit(roleChangeTask);
 110             LOG.info("RoleChangeTask submitted for execution");
 111             CheckedFuture<SetRoleOutput, RoleChangeException> taskFutureChecked = makeCheckedFuture(taskFuture);
 112             try {
 113                 SetRoleOutput setRoleOutput = taskFutureChecked.checkedGet(10, TimeUnit.SECONDS);
 114                 LOG.info("setRoleOutput received after roleChangeTask execution:{}", setRoleOutput);
 115                 resultFuture.set(RpcResultBuilder.<SetRoleOutput>success().withResult(setRoleOutput).build());
 116                 lastKnownRoleRef.set(input.getControllerRole());
 117                 return resultFuture;
 118
 119             } catch (TimeoutException | RoleChangeException e) {
 120                 roleChangeTask.incrementRetryCounter();
 121                 LOG.info("Exception in setRole(), will retry:" + (MAX_RETRIES - roleChangeTask.getRetryCounter()) + " times.", e);
 122             }
 123
 124         } while (roleChangeTask.getRetryCounter() < MAX_RETRIES);
 125
 126         resultFuture.setException(new RoleChangeException("Set Role failed after " + MAX_RETRIES + "tries on device " + input.getNode().getValue()));
 127
 128         return resultFuture;
 129     }
 130
 131     private ListenableFuture<RpcResult<SetRoleOutput>> deviceConnectionCheck() {
 132         if (!ConnectionContext.CONNECTION_STATE.WORKING.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
 133             ListenableFuture<RpcResult<SetRoleOutput>> resultingFuture = SettableFuture.create();
 134             switch (deviceContext.getPrimaryConnectionContext().getConnectionState()) {
 135                 case RIP:
 136                     final String errMsg = String.format("Device connection doesn't exist anymore. Primary connection status : %s",
 137                             deviceContext.getPrimaryConnectionContext().getConnectionState());
 138                     resultingFuture = Futures.immediateFailedFuture(new Throwable(errMsg));
 139                     break;
 140                 default:
 141                     resultingFuture = Futures.immediateCheckedFuture(RpcResultBuilder.<SetRoleOutput>failed().build());
 142                     break;
 143             }
 144             return resultingFuture;
 145         }
 146         return null;
 147     }
 148
 149     class RoleChangeTask implements Callable<SetRoleOutput> {
 150
 151         private final NodeId nodeId;
 152         private final OfpRole ofpRole;
 153         private final Short version;
 154         private final RoleService roleService;
 155         private int retryCounter = 0;
 156
 157         public RoleChangeTask(NodeId nodeId, OfpRole ofpRole, Short version, RoleService roleService) {
 158             this.nodeId = nodeId;
 159             this.ofpRole = ofpRole;
 160             this.version = version;
 161             this.roleService = roleService;
 162         }
 163
 164         @Override
 165         public SetRoleOutput call() throws RoleChangeException {
 166             LOG.info("RoleChangeTask called on device:{} OFPRole:{}", this.nodeId.getValue(), ofpRole);
 167
 168             // we cannot move ahead without having the generation id, so block the thread till we get it.
 169             BigInteger generationId = null;
 170             SetRoleOutput setRoleOutput = null;
 171
 172             try {
 173                 generationId = this.roleService.getGenerationIdFromDevice(version).get(10, TimeUnit.SECONDS);
 174                 LOG.info("RoleChangeTask, GenerationIdFromDevice from device is {}", generationId);
 175
 176             } catch (Exception e ) {
 177                 LOG.info("Exception in getting generationId for device:{}. Ex:{}" + this.nodeId.getValue(), e);
 178                 throw new RoleChangeException("Exception in getting generationId for device:"+ this.nodeId.getValue(), e);
 179             }
 180
 181
 182             LOG.info("GenerationId received from device:{} is {}", nodeId.getValue(), generationId);
 183
 184             final BigInteger nextGenerationId = getNextGenerationId(generationId);
 185
 186             LOG.info("nextGenerationId received from device:{} is {}", nodeId.getValue(), nextGenerationId);
 187
 188             try {
 189                 setRoleOutput = roleService.submitRoleChange(ofpRole, version, nextGenerationId).get(10 , TimeUnit.SECONDS);
 190                 LOG.info("setRoleOutput after submitRoleChange:{}", setRoleOutput);
 191
 192             }  catch (InterruptedException | ExecutionException |  TimeoutException e) {
 193                 LOG.error("Exception in making role change for device", e);
 194                 throw new RoleChangeException("Exception in making role change for device:" + nodeId.getValue());
 195             }
 196
 197             return setRoleOutput;
 198
 199         }
 200
 201         public void incrementRetryCounter() {
 202             this.retryCounter = retryCounter + 1;
 203         }
 204
 205         public int getRetryCounter() {
 206             return retryCounter;
 207         }
 208     }
 209
 210     public static CheckedFuture<SetRoleOutput, RoleChangeException> makeCheckedFuture(ListenableFuture<SetRoleOutput> rolePushResult) {
 211         return Futures.makeChecked(rolePushResult,
 212                 new Function<Exception, RoleChangeException>() {
 213                     @Override
 214                     public RoleChangeException apply(Exception input) {
 215                         RoleChangeException output = null;
 216                         if (input instanceof ExecutionException) {
 217                             if (input.getCause() instanceof RoleChangeException) {
 218                                 output = (RoleChangeException) input.getCause();
 219                             }
 220                         }
 221
 222                         if (output == null) {
 223                             output = new RoleChangeException(input.getMessage(), input);
 224                         }
 225
 226                         return output;
 227                     }
 228                 });
 229     }
 230 }