apps/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/actors/NetconfNodeActor.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 package org.opendaylight.netconf.topology.singleton.impl.actors;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.Props;
  12 import akka.actor.Status.Failure;
  13 import akka.actor.Status.Success;
  14 import akka.pattern.AskTimeoutException;
  15 import akka.util.Timeout;
  16 import com.google.common.base.Throwables;
  17 import com.google.common.util.concurrent.FutureCallback;
  18 import com.google.common.util.concurrent.Futures;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import com.google.common.util.concurrent.MoreExecutors;
  21 import java.io.IOException;
  22 import java.time.Duration;
  23 import java.util.List;
  24 import java.util.stream.Collectors;
  25 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor;
  26 import org.opendaylight.controller.cluster.schema.provider.impl.RemoteSchemaProvider;
  27 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
  28 import org.opendaylight.mdsal.dom.api.DOMActionResult;
  29 import org.opendaylight.mdsal.dom.api.DOMActionService;
  30 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  31 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  32 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
  33 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
  34 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  35 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  36 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  37 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  38 import org.opendaylight.netconf.client.mdsal.NetconfDevice.SchemaResourcesDTO;
  39 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
  40 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices;
  41 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Actions;
  42 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceServices.Rpcs;
  43 import org.opendaylight.netconf.dom.api.NetconfDataTreeService;
  44 import org.opendaylight.netconf.topology.singleton.impl.ProxyDOMActionService;
  45 import org.opendaylight.netconf.topology.singleton.impl.ProxyDOMRpcService;
  46 import org.opendaylight.netconf.topology.singleton.impl.ProxyYangTextSourceProvider;
  47 import org.opendaylight.netconf.topology.singleton.impl.SlaveSalFacade;
  48 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
  49 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
  50 import org.opendaylight.netconf.topology.singleton.messages.ContainerNodeMessage;
  51 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
  52 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
  53 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
  54 import org.opendaylight.netconf.topology.singleton.messages.NotMasterException;
  55 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
  56 import org.opendaylight.netconf.topology.singleton.messages.RefreshSlaveActor;
  57 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
  58 import org.opendaylight.netconf.topology.singleton.messages.UnregisterSlaveMountPoint;
  59 import org.opendaylight.netconf.topology.singleton.messages.YangTextSchemaSourceRequest;
  60 import org.opendaylight.netconf.topology.singleton.messages.action.InvokeActionMessage;
  61 import org.opendaylight.netconf.topology.singleton.messages.action.InvokeActionMessageReply;
  62 import org.opendaylight.netconf.topology.singleton.messages.netconf.NetconfDataTreeServiceRequest;
  63 import org.opendaylight.netconf.topology.singleton.messages.rpc.InvokeRpcMessage;
  64 import org.opendaylight.netconf.topology.singleton.messages.rpc.InvokeRpcMessageReply;
  65 import org.opendaylight.netconf.topology.singleton.messages.transactions.EmptyResultResponse;
  66 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewReadTransactionRequest;
  67 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewReadWriteTransactionRequest;
  68 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewWriteTransactionRequest;
  69 import org.opendaylight.yangtools.concepts.Registration;
  70 import org.opendaylight.yangtools.yang.common.QName;
  71 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  72 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  73 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  74 import org.opendaylight.yangtools.yang.model.api.source.SourceIdentifier;
  75 import org.opendaylight.yangtools.yang.model.api.source.YangTextSource;
  76 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
  77 import org.opendaylight.yangtools.yang.model.repo.api.EffectiveModelContextFactory;
  78 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
  79
  80 public class NetconfNodeActor extends AbstractUntypedActor {
  81     private final Duration writeTxIdleTimeout;
  82     private final DOMMountPointService mountPointService;
  83
  84     private SchemaResourcesDTO schemaResources;
  85     private Timeout actorResponseWaitTime;
  86     private RemoteDeviceId id;
  87     private List<SourceIdentifier> sourceIdentifiers = null;
  88     private DOMRpcService deviceRpc = null;
  89     private DOMActionService deviceAction = null;
  90     private SlaveSalFacade slaveSalManager;
  91     private DOMDataBroker deviceDataBroker;
  92     private NetconfDataTreeService netconfService;
  93     //readTxActor can be shared
  94     private ActorRef readTxActor;
  95     private List<Registration> registeredSchemas;
  96
  97     protected NetconfNodeActor(final NetconfTopologySetup setup, final RemoteDeviceId id,
  98             final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
  99         this.id = id;
 100         schemaResources = setup.getSchemaResourcesDTO();
 101         this.actorResponseWaitTime = actorResponseWaitTime;
 102         writeTxIdleTimeout = setup.getIdleTimeout();
 103         this.mountPointService = mountPointService;
 104     }
 105
 106     public static Props props(final NetconfTopologySetup setup, final RemoteDeviceId id,
 107             final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
 108         return Props.create(NetconfNodeActor.class, () ->
 109                 new NetconfNodeActor(setup, id, actorResponseWaitTime, mountPointService));
 110     }
 111
 112     @SuppressWarnings("checkstyle:IllegalCatch")
 113     @Override
 114     public void handleReceive(final Object message) {
 115         LOG.debug("{}:  received message {}", id, message);
 116
 117         if (message instanceof CreateInitialMasterActorData masterActorData) { // master
 118             sourceIdentifiers = masterActorData.getSourceIndentifiers();
 119             deviceDataBroker = masterActorData.getDeviceDataBroker();
 120             netconfService = masterActorData.getNetconfDataTreeService();
 121             final DOMDataTreeReadTransaction tx = deviceDataBroker.newReadOnlyTransaction();
 122             readTxActor = context().actorOf(ReadTransactionActor.props(tx));
 123
 124             final var deviceServices = masterActorData.getDeviceServices();
 125             deviceRpc = deviceServices.rpcs() instanceof Rpcs.Normalized normalized ? normalized.domRpcService() : null;
 126             deviceAction = deviceServices.actions() instanceof Actions.Normalized normalized ? normalized : null;
 127
 128             sender().tell(new MasterActorDataInitialized(), self());
 129             LOG.debug("{}: Master is ready.", id);
 130         } else if (message instanceof RefreshSetupMasterActorData masterActorData) {
 131             id = masterActorData.getRemoteDeviceId();
 132             schemaResources = masterActorData.getNetconfTopologyDeviceSetup().getSchemaResourcesDTO();
 133             sender().tell(new MasterActorDataInitialized(), self());
 134         } else if (message instanceof AskForMasterMountPoint askForMasterMountPoint) { // master
 135             // only master contains reference to deviceDataBroker
 136             if (deviceDataBroker != null) {
 137                 LOG.debug("{}: Sending RegisterMountPoint reply to {}", id, askForMasterMountPoint.getSlaveActorRef());
 138                 askForMasterMountPoint.getSlaveActorRef().tell(new RegisterMountPoint(sourceIdentifiers, self()),
 139                     sender());
 140             } else {
 141                 LOG.warn("{}: Received {} but we don't appear to be the master", id, askForMasterMountPoint);
 142                 sender().tell(new Failure(new NotMasterException(self())), self());
 143             }
 144         } else if (message instanceof YangTextSchemaSourceRequest yangTextSchemaSourceRequest) { // master
 145             sendYangTextSchemaSourceProxy(yangTextSchemaSourceRequest.getSourceIdentifier(), sender());
 146         } else if (message instanceof NewReadTransactionRequest) { // master
 147             sender().tell(new Success(readTxActor), self());
 148         } else if (message instanceof NewWriteTransactionRequest) { // master
 149             try {
 150                 final DOMDataTreeWriteTransaction tx = deviceDataBroker.newWriteOnlyTransaction();
 151                 final ActorRef txActor = context().actorOf(WriteTransactionActor.props(tx, writeTxIdleTimeout));
 152                 sender().tell(new Success(txActor), self());
 153             } catch (final Exception t) {
 154                 sender().tell(new Failure(t), self());
 155             }
 156         } else if (message instanceof NewReadWriteTransactionRequest) {
 157             try {
 158                 final DOMDataTreeReadWriteTransaction tx = deviceDataBroker.newReadWriteTransaction();
 159                 final ActorRef txActor = context().actorOf(ReadWriteTransactionActor.props(tx, writeTxIdleTimeout));
 160                 sender().tell(new Success(txActor), self());
 161             } catch (final Exception t) {
 162                 sender().tell(new Failure(t), self());
 163             }
 164         } else if (message instanceof InvokeRpcMessage invokeRpcMessage) { // master
 165             invokeSlaveRpc(invokeRpcMessage.getSchemaPath().lastNodeIdentifier(),
 166                 invokeRpcMessage.getNormalizedNodeMessage(), sender());
 167         } else if (message instanceof InvokeActionMessage invokeActionMessage) { // master
 168             LOG.info("InvokeActionMessage Details : {}", invokeActionMessage.toString());
 169             invokeSlaveAction(invokeActionMessage.getSchemaPath(), invokeActionMessage.getContainerNodeMessage(),
 170                 invokeActionMessage.getDOMDataTreeIdentifier(), sender());
 171         } else if (message instanceof RegisterMountPoint registerMountPoint) { //slaves
 172             sourceIdentifiers = registerMountPoint.getSourceIndentifiers();
 173             registerSlaveMountPoint(registerMountPoint.getMasterActorRef());
 174             sender().tell(new Success(null), self());
 175         } else if (message instanceof UnregisterSlaveMountPoint) { //slaves
 176             unregisterSlaveMountPoint();
 177         } else if (message instanceof RefreshSlaveActor refreshSlave) { //slave
 178             actorResponseWaitTime = refreshSlave.getActorResponseWaitTime();
 179             id = refreshSlave.getId();
 180             schemaResources = refreshSlave.getSetup().getSchemaResourcesDTO();
 181         } else if (message instanceof NetconfDataTreeServiceRequest) {
 182             final var netconfActor = context().actorOf(NetconfDataTreeServiceActor.props(netconfService,
 183                 writeTxIdleTimeout));
 184             sender().tell(new Success(netconfActor), self());
 185         }
 186     }
 187
 188     @Override
 189     public void postStop() throws Exception {
 190         try {
 191             super.postStop();
 192         } finally {
 193             unregisterSlaveMountPoint();
 194         }
 195     }
 196
 197     private void unregisterSlaveMountPoint() {
 198         if (slaveSalManager != null) {
 199             slaveSalManager.close();
 200             slaveSalManager = null;
 201         }
 202
 203         closeSchemaSourceRegistrations();
 204     }
 205
 206     private void sendYangTextSchemaSourceProxy(final SourceIdentifier sourceIdentifier, final ActorRef sender) {
 207         Futures.addCallback(
 208             schemaResources.getSchemaRepository().getSchemaSource(sourceIdentifier, YangTextSource.class),
 209             new FutureCallback<>() {
 210                 @Override
 211                 public void onSuccess(final YangTextSource yangTextSchemaSource) {
 212                     try {
 213                         LOG.debug("{}: getSchemaSource for {} succeeded", id, sourceIdentifier);
 214                         sender.tell(new YangTextSchemaSourceSerializationProxy(yangTextSchemaSource), getSelf());
 215                     } catch (IOException e) {
 216                         sender.tell(new Failure(e), getSelf());
 217                     }
 218                 }
 219
 220                 @Override
 221                 public void onFailure(final Throwable throwable) {
 222                     LOG.debug("{}: getSchemaSource for {} failed", id, sourceIdentifier, throwable);
 223                     sender.tell(new Failure(throwable), getSelf());
 224                 }
 225             }, MoreExecutors.directExecutor());
 226     }
 227
 228     private void invokeSlaveRpc(final QName qname, final NormalizedNodeMessage normalizedNodeMessage,
 229                                 final ActorRef recipient) {
 230         LOG.debug("{}: invokeSlaveRpc for {}, input: {} on rpc service {}", id, qname, normalizedNodeMessage,
 231                 deviceRpc);
 232
 233         final ListenableFuture<? extends DOMRpcResult> rpcResult = deviceRpc.invokeRpc(qname,
 234                 normalizedNodeMessage != null ? (ContainerNode) normalizedNodeMessage.getNode() : null);
 235
 236         Futures.addCallback(rpcResult, new FutureCallback<DOMRpcResult>() {
 237             @Override
 238             public void onSuccess(final DOMRpcResult domRpcResult) {
 239                 LOG.debug("{}: invokeSlaveRpc for {}, domRpcResult: {}", id, qname, domRpcResult);
 240
 241                 if (domRpcResult == null) {
 242                     recipient.tell(new EmptyResultResponse(), getSender());
 243                     return;
 244                 }
 245                 NormalizedNodeMessage nodeMessageReply = null;
 246                 if (domRpcResult.value() != null) {
 247                     nodeMessageReply = new NormalizedNodeMessage(YangInstanceIdentifier.of(), domRpcResult.value());
 248                 }
 249                 recipient.tell(new InvokeRpcMessageReply(nodeMessageReply, domRpcResult.errors()), getSelf());
 250             }
 251
 252             @Override
 253             public void onFailure(final Throwable throwable) {
 254                 recipient.tell(new Failure(throwable), getSelf());
 255             }
 256         }, MoreExecutors.directExecutor());
 257     }
 258
 259     /**
 260      * Invoking Action on Slave Node in Odl Cluster Environment.
 261      *
 262      * @param schemaPath {@link Absolute}
 263      * @param containerNodeMessage {@link ContainerNodeMessage}
 264      * @param domDataTreeIdentifier {@link DOMDataTreeIdentifier}
 265      * @param recipient {@link ActorRef}
 266      */
 267     private void invokeSlaveAction(final Absolute schemaPath, final ContainerNodeMessage containerNodeMessage,
 268             final DOMDataTreeIdentifier domDataTreeIdentifier, final ActorRef recipient) {
 269         LOG.info("{}: invokeSlaveAction for {}, input: {}, identifier: {} on action service {}", id, schemaPath,
 270             containerNodeMessage, domDataTreeIdentifier, deviceAction);
 271
 272         final ListenableFuture<? extends DOMActionResult> actionResult = deviceAction.invokeAction(schemaPath,
 273             domDataTreeIdentifier, containerNodeMessage != null ? containerNodeMessage.getNode() : null);
 274
 275         Futures.addCallback(actionResult, new FutureCallback<DOMActionResult>() {
 276
 277             @Override
 278             public void onSuccess(final DOMActionResult domActionResult) {
 279                 LOG.debug("{}: invokeSlaveAction for {}, domActionResult: {}", id, schemaPath, domActionResult);
 280                 if (domActionResult == null) {
 281                     recipient.tell(new EmptyResultResponse(), getSender());
 282                     return;
 283                 }
 284
 285                 //Check DomActionResult containing Ok onSuccess pass empty nodeMessageReply
 286                 ContainerNodeMessage nodeMessageReply = domActionResult.getOutput().map(ContainerNodeMessage::new)
 287                         .orElse(null);
 288                 recipient.tell(new InvokeActionMessageReply(nodeMessageReply, domActionResult.getErrors()), getSelf());
 289             }
 290
 291             @Override
 292             public void onFailure(final Throwable throwable) {
 293                 recipient.tell(new Failure(throwable), getSelf());
 294             }
 295         }, MoreExecutors.directExecutor());
 296     }
 297
 298     private void registerSlaveMountPoint(final ActorRef masterReference) {
 299         unregisterSlaveMountPoint();
 300
 301         slaveSalManager = new SlaveSalFacade(id, context().system(), actorResponseWaitTime, mountPointService);
 302
 303         resolveSchemaContext(createSchemaContextFactory(masterReference), slaveSalManager, masterReference, 1);
 304     }
 305
 306     private EffectiveModelContextFactory createSchemaContextFactory(final ActorRef masterReference) {
 307         final var dispatcher = getContext().dispatcher();
 308         final var remoteYangTextSourceProvider = new ProxyYangTextSourceProvider(masterReference, dispatcher,
 309             actorResponseWaitTime);
 310         final var remoteProvider = new RemoteSchemaProvider(remoteYangTextSourceProvider, dispatcher);
 311         final var schemaRegistry = schemaResources.getSchemaRegistry();
 312
 313         registeredSchemas = sourceIdentifiers.stream()
 314             .map(sourceId -> schemaRegistry.registerSchemaSource(remoteProvider, PotentialSchemaSource.create(sourceId,
 315                 YangTextSource.class, PotentialSchemaSource.Costs.REMOTE_IO.getValue())))
 316             .collect(Collectors.toList());
 317
 318         return schemaResources.getSchemaRepository().createEffectiveModelContextFactory();
 319     }
 320
 321     private void resolveSchemaContext(final EffectiveModelContextFactory schemaContextFactory,
 322             final SlaveSalFacade localSlaveSalManager, final ActorRef masterReference, final int tries) {
 323         final var schemaContextFuture = schemaContextFactory.createEffectiveModelContext(sourceIdentifiers);
 324         Futures.addCallback(schemaContextFuture, new FutureCallback<>() {
 325             @Override
 326             public void onSuccess(final EffectiveModelContext result) {
 327                 executeInSelf(() -> {
 328                     // Make sure the slaveSalManager instance hasn't changed since we initiated the schema context
 329                     // resolution.
 330                     if (slaveSalManager == localSlaveSalManager) {
 331                         LOG.info("{}: Schema context resolved: {} - registering slave mount point", id,
 332                             result.getModules());
 333                         final var actorSystem = context().system();
 334                         final var rpcProxy = new ProxyDOMRpcService(actorSystem, masterReference, id,
 335                             actorResponseWaitTime);
 336                         slaveSalManager.registerSlaveMountPoint(result, masterReference, new RemoteDeviceServices(
 337                             (Rpcs.Normalized) () -> rpcProxy,
 338                             new ProxyDOMActionService(actorSystem, masterReference, id, actorResponseWaitTime)));
 339                     }
 340                 });
 341             }
 342
 343             @Override
 344             public void onFailure(final Throwable throwable) {
 345                 executeInSelf(() -> {
 346                     if (slaveSalManager == localSlaveSalManager) {
 347                         final Throwable cause = Throwables.getRootCause(throwable);
 348                         if (cause instanceof AskTimeoutException) {
 349                             if (tries <= 5 || tries % 10 == 0) {
 350                                 LOG.warn("{}: Failed to resolve schema context - retrying...", id, throwable);
 351                             }
 352
 353                             resolveSchemaContext(schemaContextFactory, localSlaveSalManager,
 354                                     masterReference, tries + 1);
 355                         } else {
 356                             LOG.error("{}: Failed to resolve schema context - unable to register slave mount point",
 357                                     id, throwable);
 358                             closeSchemaSourceRegistrations();
 359                         }
 360                     }
 361                 });
 362             }
 363         }, MoreExecutors.directExecutor());
 364     }
 365
 366     private void closeSchemaSourceRegistrations() {
 367         if (registeredSchemas != null) {
 368             registeredSchemas.forEach(Registration::close);
 369             registeredSchemas = null;
 370         }
 371     }
 372 }