2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.netconf.topology.singleton.impl.actors;
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;
80 public class NetconfNodeActor extends AbstractUntypedActor {
81 private final Duration writeTxIdleTimeout;
82 private final DOMMountPointService mountPointService;
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;
97 protected NetconfNodeActor(final NetconfTopologySetup setup, final RemoteDeviceId id,
98 final Timeout actorResponseWaitTime, final DOMMountPointService mountPointService) {
100 schemaResources = setup.getSchemaResourcesDTO();
101 this.actorResponseWaitTime = actorResponseWaitTime;
102 writeTxIdleTimeout = setup.getIdleTimeout();
103 this.mountPointService = mountPointService;
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));
112 @SuppressWarnings("checkstyle:IllegalCatch")
114 public void handleReceive(final Object message) {
115 LOG.debug("{}: received message {}", id, message);
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));
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;
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()),
141 LOG.warn("{}: Received {} but we don't appear to be the master", id, askForMasterMountPoint);
142 sender().tell(new Failure(new NotMasterException(self())), self());
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
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());
156 } else if (message instanceof NewReadWriteTransactionRequest) {
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());
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());
189 public void postStop() throws Exception {
193 unregisterSlaveMountPoint();
197 private void unregisterSlaveMountPoint() {
198 if (slaveSalManager != null) {
199 slaveSalManager.close();
200 slaveSalManager = null;
203 closeSchemaSourceRegistrations();
206 private void sendYangTextSchemaSourceProxy(final SourceIdentifier sourceIdentifier, final ActorRef sender) {
208 schemaResources.getSchemaRepository().getSchemaSource(sourceIdentifier, YangTextSource.class),
209 new FutureCallback<>() {
211 public void onSuccess(final YangTextSource yangTextSchemaSource) {
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());
221 public void onFailure(final Throwable throwable) {
222 LOG.debug("{}: getSchemaSource for {} failed", id, sourceIdentifier, throwable);
223 sender.tell(new Failure(throwable), getSelf());
225 }, MoreExecutors.directExecutor());
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,
233 final ListenableFuture<? extends DOMRpcResult> rpcResult = deviceRpc.invokeRpc(qname,
234 normalizedNodeMessage != null ? (ContainerNode) normalizedNodeMessage.getNode() : null);
236 Futures.addCallback(rpcResult, new FutureCallback<DOMRpcResult>() {
238 public void onSuccess(final DOMRpcResult domRpcResult) {
239 LOG.debug("{}: invokeSlaveRpc for {}, domRpcResult: {}", id, qname, domRpcResult);
241 if (domRpcResult == null) {
242 recipient.tell(new EmptyResultResponse(), getSender());
245 NormalizedNodeMessage nodeMessageReply = null;
246 if (domRpcResult.value() != null) {
247 nodeMessageReply = new NormalizedNodeMessage(YangInstanceIdentifier.of(), domRpcResult.value());
249 recipient.tell(new InvokeRpcMessageReply(nodeMessageReply, domRpcResult.errors()), getSelf());
253 public void onFailure(final Throwable throwable) {
254 recipient.tell(new Failure(throwable), getSelf());
256 }, MoreExecutors.directExecutor());
260 * Invoking Action on Slave Node in Odl Cluster Environment.
262 * @param schemaPath {@link Absolute}
263 * @param containerNodeMessage {@link ContainerNodeMessage}
264 * @param domDataTreeIdentifier {@link DOMDataTreeIdentifier}
265 * @param recipient {@link ActorRef}
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);
272 final ListenableFuture<? extends DOMActionResult> actionResult = deviceAction.invokeAction(schemaPath,
273 domDataTreeIdentifier, containerNodeMessage != null ? containerNodeMessage.getNode() : null);
275 Futures.addCallback(actionResult, new FutureCallback<DOMActionResult>() {
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());
285 //Check DomActionResult containing Ok onSuccess pass empty nodeMessageReply
286 ContainerNodeMessage nodeMessageReply = domActionResult.getOutput().map(ContainerNodeMessage::new)
288 recipient.tell(new InvokeActionMessageReply(nodeMessageReply, domActionResult.getErrors()), getSelf());
292 public void onFailure(final Throwable throwable) {
293 recipient.tell(new Failure(throwable), getSelf());
295 }, MoreExecutors.directExecutor());
298 private void registerSlaveMountPoint(final ActorRef masterReference) {
299 unregisterSlaveMountPoint();
301 slaveSalManager = new SlaveSalFacade(id, context().system(), actorResponseWaitTime, mountPointService);
303 resolveSchemaContext(createSchemaContextFactory(masterReference), slaveSalManager, masterReference, 1);
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();
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());
318 return schemaResources.getSchemaRepository().createEffectiveModelContextFactory();
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<>() {
326 public void onSuccess(final EffectiveModelContext result) {
327 executeInSelf(() -> {
328 // Make sure the slaveSalManager instance hasn't changed since we initiated the schema context
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)));
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);
353 resolveSchemaContext(schemaContextFactory, localSlaveSalManager,
354 masterReference, tries + 1);
356 LOG.error("{}: Failed to resolve schema context - unable to register slave mount point",
358 closeSchemaSourceRegistrations();
363 }, MoreExecutors.directExecutor());
366 private void closeSchemaSourceRegistrations() {
367 if (registeredSchemas != null) {
368 registeredSchemas.forEach(Registration::close);
369 registeredSchemas = null;