path: root/sdnr/wt/devicemanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/devicemanager/impl/DeviceManagerImpl.java

/*******************************************************************************
 * ============LICENSE_START========================================================================
 * ONAP : ccsdk feature sdnr wt
 * =================================================================================================
 * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
 * =================================================================================================
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 * ============LICENSE_END==========================================================================
 ******************************************************************************/
package org.onap.ccsdk.features.sdnr.wt.devicemanager.impl;

import com.google.common.base.Optional;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.aaiconnector.impl.AaiProviderClient;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.archiveservice.ArchiveCleanService;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.database.HtDatabaseNode;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.netconf.ONFCoreNetworkElementFactory;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.netconf.ONFCoreNetworkElementRepresentation;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.toggleAlarmFilter.NotificationDelayService;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.HtDevicemanagerConfiguration;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.AkkaConfig;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.EsConfig;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.GeoConfig;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.PmConfig;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.dcaeconnector.impl.DcaeProviderClient;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.devicemonitor.impl.DeviceMonitor;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.devicemonitor.impl.DeviceMonitorEmptyImpl;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.devicemonitor.impl.DeviceMonitorImpl;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.database.service.HtDatabaseEventsService;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.listener.NetconfChangeListener;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.listener.ODLEventListener;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.ProblemNotificationXml;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.WebSocketServiceClient;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.WebSocketServiceClientDummyImpl;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.WebSocketServiceClientImpl2;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.index.impl.IndexConfigService;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.index.impl.IndexMwtnService;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.index.impl.IndexUpdateService;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.maintenance.impl.MaintenanceServiceImpl;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.performancemanager.impl.PerformanceManagerImpl;
import org.onap.ccsdk.features.sdnr.wt.devicemanager.performancemanager.impl.database.service.MicrowaveHistoricalPerformanceWriterService;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.md.sal.binding.api.MountPoint;
import org.opendaylight.controller.md.sal.binding.api.MountPointService;
import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInputBuilder;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.NotificationsService;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.ClusteredConnectionStatus;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.network.topology.topology.topology.types.TopologyNetconf;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;

@SuppressWarnings("deprecation")
public class DeviceManagerImpl implements DeviceManagerService, AutoCloseable, ResyncNetworkElementsListener {

    private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
    private static final String APPLICATION_NAME = "DeviceManager";
    private static final String MYDBKEYNAMEBASE = "SDN-Controller";

    // http://sendateodl:8181/restconf/operational/network-topology:network-topology/topology/topology-netconf
    private static final InstanceIdentifier<Topology> NETCONF_TOPO_IID =
            InstanceIdentifier.create(NetworkTopology.class).child(Topology.class,
                    new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())));
    @SuppressWarnings("unused")
    private static final String STARTUPLOG_FILENAME = "etc/devicemanager.startup.log";
    // private static final String STARTUPLOG_FILENAME2 = "data/cache/devicemanager.startup.log";

    private DataBroker dataBroker = null;
    private MountPointService mountPointService = null;
    private RpcProviderRegistry rpcProviderRegistry = null;
    @SuppressWarnings("unused")
    private NotificationPublishService notificationPublishService = null;
    private ClusterSingletonServiceProvider clusterSingletonServiceProvider;

    private final ConcurrentHashMap<String, ONFCoreNetworkElementRepresentation> networkElementRepresentations =
            new ConcurrentHashMap<>();

    private WebSocketServiceClient webSocketService;
    private HtDatabaseEventsService databaseClientEvents;
    private ODLEventListener odlEventListener;
    private NetconfChangeListener netconfChangeListener;
    private DeviceManagerApiServiceImpl rpcApiService;
    private @Nullable PerformanceManagerImpl performanceManager = null;
    private ProviderClient dcaeProviderClient;
    private ProviderClient aotsMProvider;
    private @Nullable AaiProviderClient aaiProviderClient = null;
    private @Nullable DeviceMonitor deviceMonitor = new DeviceMonitorEmptyImpl();
    private IndexUpdateService updateService;
    private IndexConfigService configService;
    private IndexMwtnService mwtnService;
    private HtDatabaseNode htDatabase;
    private Boolean devicemanagerInitializationOk = false;
    private MaintenanceServiceImpl maintenanceService;
    private NotificationDelayService<ProblemNotificationXml> notificationDelayService;
    private Thread threadDoClearCurrentFaultByNodename = null;
    private int refreshCounter = 0;
    private AkkaConfig akkaConfig;
    private ArchiveCleanService archiveCleanService;
    @SuppressWarnings("unused")
	private ClusterSingletonServiceRegistration cssRegistration;

    // Blueprint 1
    public DeviceManagerImpl() {
        LOG.info("Creating provider for {}", APPLICATION_NAME);
    }

    public void setDataBroker(DataBroker dataBroker) {
        this.dataBroker = dataBroker;
    }

    public void setRpcProviderRegistry(RpcProviderRegistry rpcProviderRegistry) {
        this.rpcProviderRegistry = rpcProviderRegistry;
    }

    public void setNotificationPublishService(NotificationPublishService notificationPublishService) {
        this.notificationPublishService = notificationPublishService;
    }

    public void setMountPointService(MountPointService mountPointService) {
        this.mountPointService = mountPointService;
    }
    public void setClusterSingletonService(ClusterSingletonServiceProvider clusterSingletonService) {
    	this.clusterSingletonServiceProvider = clusterSingletonService;
    }
    public void init() {

        LOG.info("Session Initiated start {}", APPLICATION_NAME);

        // Start RPC Service
        this.rpcApiService = new DeviceManagerApiServiceImpl(rpcProviderRegistry);
        // Get configuration
        HtDevicemanagerConfiguration config = HtDevicemanagerConfiguration.getConfiguration();
        try {
            this.akkaConfig = AkkaConfig.load();
            LOG.debug("akka.conf loaded: " + akkaConfig.toString());
        } catch (Exception e1) {
            this.akkaConfig = null;
            LOG.warn("problem loading akka.conf: " + e1.getMessage());
        }
        GeoConfig geoConfig = null;
        if (akkaConfig != null && akkaConfig.isCluster()) {
            LOG.info("cluster mode detected");
            if (GeoConfig.fileExists()) {
                try {
                    LOG.debug("try to load geoconfig");
                    geoConfig = GeoConfig.load();
                } catch (Exception err) {
                    LOG.warn("problem loading geoconfig: " + err.getMessage());
                }
            } else {
                LOG.debug("no geoconfig file found");
            }
        } else {
            LOG.info("single node mode detected");
        }

        this.notificationDelayService = new NotificationDelayService<>(config);

        EsConfig dbConfig = config.getEs();
        LOG.debug("esConfig=" + dbConfig.toString());
        // Start database
        htDatabase = HtDatabaseNode.start(dbConfig, akkaConfig, geoConfig);
        if (htDatabase == null) {
            LOG.error("Can only run with local database. Stop initialization of devicemanager.");
        } else {
            // init Database Values only if singleNode or clusterMember=1
            if (akkaConfig == null || akkaConfig.isClusterAndFirstNode()) {
                // Create DB index if not existing and if database is running
                try {
                    this.configService = new IndexConfigService(htDatabase);
                    this.mwtnService = new IndexMwtnService(htDatabase);
                } catch (Exception e) {
                    LOG.warn("Can not start ES access clients to provide database index config, mwtn. ", e);
                }
            }
            // start service for device maintenance service
            this.maintenanceService = new MaintenanceServiceImpl(htDatabase);

            // Websockets
            try {
                this.webSocketService = new WebSocketServiceClientImpl2(rpcProviderRegistry);
            } catch (Exception e) {
                LOG.error("Can not start websocket service. Loading mock class.", e);
                this.webSocketService = new WebSocketServiceClientDummyImpl();
            }
            // DCAE
            this.dcaeProviderClient = new DcaeProviderClient(config, dbConfig.getCluster(), this);

            this.aaiProviderClient = new AaiProviderClient(config, this);
            // EM
            EsConfig emConfig = dbConfig.cloneWithIndex("sdnevents");

            if (emConfig == null) {
                LOG.warn("No configuration available. Don't start event manager");
            } else {
                this.databaseClientEvents = new HtDatabaseEventsService(htDatabase);

                String myDbKeyNameExtended = MYDBKEYNAMEBASE + "-" + dbConfig.getCluster();

                this.odlEventListener = new ODLEventListener(myDbKeyNameExtended, webSocketService,
                        databaseClientEvents, dcaeProviderClient, aotsMProvider, maintenanceService);
            }
            this.archiveCleanService = new ArchiveCleanService(config, databaseClientEvents, mwtnService);
            this.cssRegistration = this.clusterSingletonServiceProvider.registerClusterSingletonService(this.archiveCleanService);
            // PM
            PmConfig configurationPM = config.getPm();
            LOG.info("Performance manager configuration: {}", configurationPM);
            if (!configurationPM.isPerformanceManagerEnabled()) {

                LOG.info("No configuration available. Don't start performance manager");
            } else {
                @Nullable
                MicrowaveHistoricalPerformanceWriterService databaseClientHistoricalPerformance;
                databaseClientHistoricalPerformance = new MicrowaveHistoricalPerformanceWriterService(htDatabase);
                this.performanceManager = new PerformanceManagerImpl(60, databaseClientHistoricalPerformance);
            }

            // DUS (Database update service)
            LOG.debug("start db update service");
            this.updateService =
                    new IndexUpdateService(htDatabase, dbConfig.getHost(), dbConfig.getCluster(), dbConfig.getNode());
            this.updateService.start();

            // RPC Service for specific services
            this.rpcApiService.setMaintenanceService(this.maintenanceService);
            this.rpcApiService.setResyncListener(this);
            // DM
            // DeviceMonitor has to be available before netconfSubscriptionManager is
            // configured
            LOG.debug("start DeviceMonitor Service");
            this.deviceMonitor = new DeviceMonitorImpl(dataBroker, odlEventListener, config);

            // netconfSubscriptionManager should be the last one because this is a callback
            // service
            LOG.debug("start NetconfSubscriptionManager Service");
            // this.netconfSubscriptionManager = new
            // NetconfSubscriptionManagerOfDeviceManager(this, dataBroker);
            // this.netconfSubscriptionManager.register();
            this.netconfChangeListener = new NetconfChangeListener(this, dataBroker);
            this.netconfChangeListener.register();

            this.devicemanagerInitializationOk = true;
        }
        LOG.info("Session Initiated end. Initialization done {}", devicemanagerInitializationOk);
    }

    @Override
    public void close() throws Exception {
        LOG.info("DeviceManagerImpl closing ...");

        close(performanceManager);
        close(dcaeProviderClient);
        close(aaiProviderClient);
        close(aotsMProvider);
        close(deviceMonitor);
        close(updateService, configService, mwtnService);
        close(htDatabase);
        close(netconfChangeListener);
        close(maintenanceService);
        close(rpcApiService);
        close(notificationDelayService);
        close(archiveCleanService);
        LOG.info("DeviceManagerImpl closing done");
    }


    /**
     * Used to close all Services, that should support AutoCloseable Pattern
     *
     * @param toClose
     * @throws Exception
     */
    private void close(AutoCloseable... toCloseList) throws Exception {
        for (AutoCloseable element : toCloseList) {
            if (element != null) {
                element.close();
            }
        }
    }

    /*-------------------------------------------------------------------------------------------
     * Functions for interface DeviceManagerService
     */

    /**
     * For each mounted device a mountpoint is created and this listener is called.
     * Mountpoint was created or existing. Managed device is now fully connected to node/mountpoint.
     * @param action provide action
     * @param nNodeId id of the mountpoint
     * @param nNode mountpoint contents
     */
    public void startListenerOnNodeForConnectedState(Action action, NodeId nNodeId, NetconfNode nNode) {

        String mountPointNodeName = nNodeId.getValue();
        LOG.info("Starting Event listener on Netconf for mountpoint {} Action {}", mountPointNodeName, action);

        boolean preConditionMissing = false;
        if (mountPointService == null) {
            preConditionMissing = true;
            LOG.warn("No mountservice available.");
        }
        if (!devicemanagerInitializationOk) {
            preConditionMissing = true;
            LOG.warn("Devicemanager initialization still pending.");
        }
        if (preConditionMissing) {
            return;
        }

        if (networkElementRepresentations.containsKey(mountPointNodeName)) {
            LOG.warn("Mountpoint {} already registered. Leave startup procedure.", mountPointNodeName);
            return;
        }

        if (!isNetconfNodeMaster(nNode)) {
            // Change Devicemonitor-status to connected ... for non master mountpoints.
            deviceMonitor.deviceConnectSlaveIndication(mountPointNodeName);
            return;
        }

        InstanceIdentifier<Node> instanceIdentifier =
                NETCONF_TOPO_IID.child(Node.class, new NodeKey(new NodeId(mountPointNodeName)));

        Optional<MountPoint> optionalMountPoint = null;
        int timeout = 10000;
        while (!(optionalMountPoint = mountPointService.getMountPoint(instanceIdentifier)).isPresent() && timeout > 0) {
            LOG.info("Event listener waiting for mount point for Netconf device :: Name : {}", mountPointNodeName);
            sleepMs(1000);
            timeout -= 1000;
        }

        if (!optionalMountPoint.isPresent()) {
            LOG.warn("Event listener timeout while waiting for mount point for Netconf device :: Name : {} ",
                    mountPointNodeName);
            return;
        }
        // Mountpoint is present for sure
        MountPoint mountPoint = optionalMountPoint.get();
        //BindingDOMDataBrokerAdapter.BUILDER_FACTORY;
        LOG.info("Mountpoint with id: {} class {} toString {}", mountPoint.getIdentifier(), mountPoint.getClass().getName(), mountPoint);
        Optional<DataBroker> optionalNetconfNodeDatabroker = mountPoint.getService(DataBroker.class);

        if (! optionalNetconfNodeDatabroker.isPresent()) {
            LOG.info("Slave mountpoint {} without databroker", mountPointNodeName);
            return;
        }

        DataBroker netconfNodeDataBroker = optionalNetconfNodeDatabroker.get();
        LOG.info("Master mountpoint {}", mountPointNodeName);
        // getNodeInfoTest(dataBroker);

        // create automatic empty maintenance entry into db before reading and listening for problems
        this.maintenanceService.createIfNotExists(mountPointNodeName);

        // Setup microwaveEventListener for Notificationservice

        // MicrowaveEventListener microwaveEventListener = new
        // MicrowaveEventListener(mountPointNodeName, websocketmanagerService,
        // xmlMapper, databaseClientEvents);
        ONFCoreNetworkElementRepresentation ne = ONFCoreNetworkElementFactory.create(mountPointNodeName, dataBroker,
                webSocketService, databaseClientEvents, instanceIdentifier, netconfNodeDataBroker, dcaeProviderClient,
                aotsMProvider, maintenanceService, notificationDelayService);
        networkElementRepresentations.put(mountPointNodeName, ne);
        ne.doRegisterMicrowaveEventListener(mountPoint);

        // Register netconf stream
        registerNotificationStream(mountPointNodeName, mountPoint, "NETCONF");

        // -- Read data from NE
        ne.initialReadFromNetworkElement();
        ne.initSynchronizationExtension();


        sendUpdateNotification(mountPointNodeName, nNode.getConnectionStatus());

        if (aaiProviderClient != null) {
            aaiProviderClient.onDeviceRegistered(mountPointNodeName);
        }
        // -- Register NE to performance manager
        if (performanceManager != null) {
            performanceManager.registration(mountPointNodeName, ne);
        }

        deviceMonitor.deviceConnectMasterIndication(mountPointNodeName, ne);

        LOG.info("Starting Event listener on Netconf device :: Name : {} finished", mountPointNodeName);
    }

    /**
     * Mountpoint created or existing. Managed device is actually disconnected from node/ mountpoint.
     * Origin state: Connecting, Connected
     * Target state: are UnableToConnect or Connecting
     * @param action create or update
     * @param nNodeId id of the mountpoint
     * @param nNode mountpoint contents
     */
    public void enterNonConnectedState(Action action, NodeId nNodeId, NetconfNode nNode) {
        String mountPointNodeName = nNodeId.getValue();
        ConnectionStatus csts = nNode.getConnectionStatus();
        if (isNetconfNodeMaster(nNode)) {
        	sendUpdateNotification(mountPointNodeName, csts);
        }

        // Handling if mountpoint exist. connected -> connecting/UnableToConnect
        stopListenerOnNodeForConnectedState(mountPointNodeName);

        deviceMonitor.deviceDisconnectIndication(mountPointNodeName);

    }

    /**
     * Mountpoint removed indication.
     * @param nNodeId id of the mountpoint
     */
    public void removeMountpointState(NodeId nNodeId) {
        String mountPointNodeName = nNodeId.getValue();
        LOG.info("mountpointNodeRemoved {}", nNodeId.getValue());

        stopListenerOnNodeForConnectedState(mountPointNodeName);
        deviceMonitor.removeMountpointIndication(mountPointNodeName);
        if (odlEventListener != null) {
            odlEventListener.deRegistration(mountPointNodeName);
        }
    }

    /**
     * Do all tasks necessary to move from mountpoint state connected -> connecting
     * @param mountPointNodeName provided
     * @param ne representing the device connected to mountpoint
     */
    private void stopListenerOnNodeForConnectedState( String mountPointNodeName) {
        ONFCoreNetworkElementRepresentation ne = networkElementRepresentations.remove(mountPointNodeName);
        if (ne != null) {
            this.maintenanceService.deleteIfNotRequired(mountPointNodeName);
            int problems = ne.removeAllCurrentProblemsOfNode();
            LOG.debug("Removed all {} problems from database at deregistration for {}", problems, mountPointNodeName);
            if (performanceManager != null) {
                performanceManager.deRegistration(mountPointNodeName);
            }
            if (aaiProviderClient != null) {
                aaiProviderClient.onDeviceUnregistered(mountPointNodeName);
            }
        }
    }

    private void sendUpdateNotification(String mountPointNodeName, ConnectionStatus csts) {
        LOG.info("enter Non ConnectedState for device :: Name : {} ConnectionStatus {}", mountPointNodeName, csts);
        if (odlEventListener != null) {
        	odlEventListener.updateRegistration(mountPointNodeName, ConnectionStatus.class.getSimpleName(), csts != null ? csts.getName() : "null");
        }
    }

    /**
     * Handle netconf/mountpoint changes
     */
    @Override
    public void netconfChangeHandler(Action action, @Nullable ConnectionStatus csts, NodeId nodeId, NetconfNode nNode) {

		ClusteredConnectionStatus ccsts = nNode.getClusteredConnectionStatus();
		String nodeIdString = nodeId.getValue();
		if (action == Action.CREATE) {
	        if (odlEventListener != null) {
	        	odlEventListener.registration(nodeIdString);
	        }
		}
		boolean isCluster = akkaConfig == null && akkaConfig.isCluster();
		if (isCluster && ccsts == null) {
			LOG.debug("NETCONF Node {} {} does not provide cluster status. Stop execution.", nodeIdString, action);
		} else {
			switch (action) {
			case REMOVE:
				removeMountpointState(nodeId); // Stop Monitor
				break;

			case UPDATE:
			case CREATE:
				if (csts != null) {
					switch (csts) {
					case Connected: {
						startListenerOnNodeForConnectedState(action, nodeId, nNode);
						break;
					}
					case UnableToConnect:
					case Connecting: {
						enterNonConnectedState(action, nodeId, nNode);
						break;
					}
					}
				} else {
					LOG.debug("NETCONF Node handled with null status for action", action);
				}
				break;
			}
		}
    }

    /*-------------------------------------------------------------------------------------------
     * Functions
     */

    public ArchiveCleanService getArchiveCleanService() {
        return this.archiveCleanService;
    }

    public HtDatabaseEventsService getDatabaseClientEvents() {
        return databaseClientEvents;
    }

    public IndexMwtnService getMwtnService() {
        return mwtnService;
    }

    /**
     * Async RPC Interface implementation
     */
    @Override
    public @Nonnull List<String> doClearCurrentFaultByNodename(@Nullable List<String> nodeNamesInput)
            throws IllegalStateException {

        if (this.databaseClientEvents == null) {
            throw new IllegalStateException("dbEvents service not instantiated");
        }

        if (threadDoClearCurrentFaultByNodename != null && threadDoClearCurrentFaultByNodename.isAlive()) {
            throw new IllegalStateException("A clear task is already active");
        } else {

            // Create list of mountpoints if input is empty, using the content in ES
            if (nodeNamesInput == null || nodeNamesInput.size() <= 0) {
                nodeNamesInput = this.databaseClientEvents.getAllNodesWithCurrentAlarms();
            }

            // Filter all mountpoints from input that were found and are known to this Cluster-node instance of
            // DeviceManager
            final List<String> nodeNamesHandled = new ArrayList<>();
            for (String mountpointName : nodeNamesInput) {
                LOG.info("Work with mountpoint {}", mountpointName);

                if (odlEventListener != null && mountpointName.equals(odlEventListener.getOwnKeyName())) {

                    // SDN Controller related alarms
                    // -- can not be recreated on all nodes in connected state
                    // -- would result in a DCAE/AAI Notification
                    // Conclusion for 1810 Delivery ... not covered by RPC function (See issue #43)
                    LOG.info("Ignore SDN Controller related alarms for {}", mountpointName);
                    // this.databaseClientEvents.clearFaultsCurrentOfNode(mountpointName);
                    // nodeNamesHandled.add(mountpointName);

                } else {

                    if (mountPointService != null) {
                        InstanceIdentifier<Node> instanceIdentifier =
                                NETCONF_TOPO_IID.child(Node.class, new NodeKey(new NodeId(mountpointName)));
                        Optional<MountPoint> optionalMountPoint = mountPointService.getMountPoint(instanceIdentifier);

                        if (!optionalMountPoint.isPresent()) {
                            LOG.info("Remove Alarms for unknown mountpoint {}", mountpointName);
                            this.databaseClientEvents.clearFaultsCurrentOfNode(mountpointName);
                            nodeNamesHandled.add(mountpointName);
                        } else {
                            if (networkElementRepresentations.containsKey(mountpointName)) {
                                LOG.info("At node known mountpoint {}", mountpointName);
                                nodeNamesHandled.add(mountpointName);
                            } else {
                                LOG.info("At node unknown mountpoint {}", mountpointName);
                            }
                        }
                    }
                }
            }

            // Force a sync
            this.deviceMonitor.refreshAlarmsInDb();

            threadDoClearCurrentFaultByNodename = new Thread(() -> {
                refreshCounter++;
                LOG.info("Start refresh mountpoint task {}", refreshCounter);
                // for(String nodeName:nodeNamesOutput) {
                for (String nodeName : nodeNamesHandled) {
                    ONFCoreNetworkElementRepresentation ne = networkElementRepresentations.get(nodeName);
                    if (ne != null) {
                        LOG.info("Refresh mountpoint {}", nodeName);
                        ne.initialReadFromNetworkElement();
                    } else {
                        LOG.info("Unhandled mountpoint {}", nodeName);
                    }
                }
                LOG.info("End refresh mountpoint task {}", refreshCounter);
            });
            threadDoClearCurrentFaultByNodename.start();
            return nodeNamesHandled;
        }
    };

    /**
     * Indication if init() of devicemanager successfully done.
     *
     * @return true if init() was sucessfull. False if not done or not successfull.
     */
    public boolean isDevicemanagerInitializationOk() {
        return this.devicemanagerInitializationOk;
    }

    /**
     * Get initialization status of database.
     *
     * @return true if fully initialized false if not
     */
    public boolean isDatabaseInitializationFinished() {
        return htDatabase == null ? false : htDatabase.getInitialized();
    }

    /*---------------------------------------------------------------------
     * Private funtions
     */

    /**
     * Do the stream creation for the device.
     *
     * @param mountPointNodeName
     * @param mountPoint
     */
    private void registerNotificationStream(String mountPointNodeName, MountPoint mountPoint, String streamName) {

        final Optional<RpcConsumerRegistry> optionalRpcConsumerService =
                mountPoint.getService(RpcConsumerRegistry.class);
        if (optionalRpcConsumerService.isPresent()) {
            final RpcConsumerRegistry rpcConsumerRegitry = optionalRpcConsumerService.get();
            final NotificationsService rpcService = rpcConsumerRegitry.getRpcService(NotificationsService.class);
            if (rpcService == null) {
                LOG.warn("rpcService is null for mountpoint {}", mountPointNodeName);
            } else {
                final CreateSubscriptionInputBuilder createSubscriptionInputBuilder =
                        new CreateSubscriptionInputBuilder();
                createSubscriptionInputBuilder.setStream(new StreamNameType(streamName));
                LOG.info("Event listener triggering notification stream {} for node {}", streamName,
                        mountPointNodeName);
                try {
                    CreateSubscriptionInput createSubscriptionInput = createSubscriptionInputBuilder.build();
                    if (createSubscriptionInput == null) {
                        LOG.warn("createSubscriptionInput is null for mountpoint {}", mountPointNodeName);
                    } else {
                        rpcService.createSubscription(createSubscriptionInput);
                    }
                } catch (NullPointerException e) {
                    LOG.warn("createSubscription failed");
                }
            }
        } else {
            LOG.warn("No RpcConsumerRegistry avaialble.");
        }

    }

    /**
     * Get NE object
     *
     * @param mountpoint mount point name
     * @return null or NE specific data
     */
    public @Nullable ONFCoreNetworkElementRepresentation getNeByMountpoint(String mountpoint) {

        return networkElementRepresentations.get(mountpoint);

    }

    /* -- LOG related functions -- */


    private boolean isInClusterMode() {
        return this.akkaConfig == null ? false : this.akkaConfig.isCluster();
    }

    private String getClusterNetconfNodeName() {
        return this.akkaConfig == null ? "" : this.akkaConfig.getClusterConfig().getClusterSeedNodeName("abc");
    }

    private boolean isNetconfNodeMaster(NetconfNode nnode) {
        if (isInClusterMode()) {
            LOG.debug("check if me is responsible for node");
            String masterNodeName = nnode.getClusteredConnectionStatus() == null ? "null"
                    : nnode.getClusteredConnectionStatus().getNetconfMasterNode();
            String myNodeName = getClusterNetconfNodeName();
            LOG.debug("sdnMasterNode=" + masterNodeName + " and sdnMyNode=" + myNodeName);
            if (!masterNodeName.equals(myNodeName)) {
                LOG.debug("netconf change but me is not master for this node");
                return false;
            }
        }
        return true;
    }


    private void sleepMs(int milliseconds) {
        try {
            Thread.sleep(milliseconds);
        } catch (InterruptedException e) {
            LOG.debug("Interrupted sleep");
            // Restore interrupted state...
            Thread.currentThread().interrupt();
        }
    }

}