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

   1 /*******************************************************************************
   2  * ============LICENSE_START========================================================================
   3  * ONAP : ccsdk feature sdnr wt
   4  * =================================================================================================
   5  * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
   6  * =================================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   8  * in compliance with the License. You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software distributed under the License
  13  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  14  * or implied. See the License for the specific language governing permissions and limitations under
  15  * the License.
  16  * ============LICENSE_END==========================================================================
  17  ******************************************************************************/
  18 package org.onap.ccsdk.features.sdnr.wt.devicemanager.impl;
  19
  20 import com.google.common.base.Optional;
  21 import java.util.ArrayList;
  22 import java.util.List;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import javax.annotation.Nonnull;
  25 import javax.annotation.Nullable;
  26 import org.onap.ccsdk.features.sdnr.wt.devicemanager.aaiconnector.impl.AaiProviderClient;
  27 import org.onap.ccsdk.features.sdnr.wt.devicemanager.archiveservice.ArchiveCleanService;
  28 import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.database.HtDatabaseNode;
  29 import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.netconf.ONFCoreNetworkElementFactory;
  30 import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.netconf.ONFCoreNetworkElementRepresentation;
  31 import org.onap.ccsdk.features.sdnr.wt.devicemanager.base.toggleAlarmFilter.NotificationDelayService;
  32 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.HtDevicemanagerConfiguration;
  33 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.AkkaConfig;
  34 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.EsConfig;
  35 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.GeoConfig;
  36 import org.onap.ccsdk.features.sdnr.wt.devicemanager.config.impl.PmConfig;
  37 import org.onap.ccsdk.features.sdnr.wt.devicemanager.dcaeconnector.impl.DcaeProviderClient;
  38 import org.onap.ccsdk.features.sdnr.wt.devicemanager.devicemonitor.impl.DeviceMonitorImpl;
  39 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.database.service.HtDatabaseEventsService;
  40 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.listener.NetconfChangeListener;
  41 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.listener.ODLEventListener;
  42 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.ProblemNotificationXml;
  43 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.WebSocketServiceClient;
  44 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.WebSocketServiceClientDummyImpl;
  45 import org.onap.ccsdk.features.sdnr.wt.devicemanager.impl.xml.WebSocketServiceClientImpl2;
  46 import org.onap.ccsdk.features.sdnr.wt.devicemanager.index.impl.IndexConfigService;
  47 import org.onap.ccsdk.features.sdnr.wt.devicemanager.index.impl.IndexMwtnService;
  48 import org.onap.ccsdk.features.sdnr.wt.devicemanager.index.impl.IndexUpdateService;
  49 import org.onap.ccsdk.features.sdnr.wt.devicemanager.maintenance.impl.MaintenanceServiceImpl;
  50 import org.onap.ccsdk.features.sdnr.wt.devicemanager.performancemanager.impl.PerformanceManagerImpl;
  51 import org.onap.ccsdk.features.sdnr.wt.devicemanager.performancemanager.impl.database.service.MicrowaveHistoricalPerformanceWriterService;
  52 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  53 import org.opendaylight.controller.md.sal.binding.api.MountPoint;
  54 import org.opendaylight.controller.md.sal.binding.api.MountPointService;
  55 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  56 import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
  57 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  58 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
  59 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInputBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.NotificationsService;
  61 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.network.topology.topology.topology.types.TopologyNetconf;
  65 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  66 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  67 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  68 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  69 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  70 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  71 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  72 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  73 import org.slf4j.Logger;
  74 import org.slf4j.LoggerFactory;
  75 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  76 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  77
  78 @SuppressWarnings("deprecation")
  79 public class DeviceManagerImpl implements DeviceManagerService, AutoCloseable, ResyncNetworkElementsListener {
  80
  81     private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
  82     private static final String APPLICATION_NAME = "DeviceManager";
  83     private static final String MYDBKEYNAMEBASE = "SDN-Controller";
  84
  85     // http://sendateodl:8181/restconf/operational/network-topology:network-topology/topology/topology-netconf
  86     private static final InstanceIdentifier<Topology> NETCONF_TOPO_IID =
  87             InstanceIdentifier.create(NetworkTopology.class).child(Topology.class,
  88                     new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())));
  89     @SuppressWarnings("unused")
  90     private static final String STARTUPLOG_FILENAME = "etc/devicemanager.startup.log";
  91     // private static final String STARTUPLOG_FILENAME2 = "data/cache/devicemanager.startup.log";
  92
  93     private DataBroker dataBroker = null;
  94     private MountPointService mountPointService = null;
  95     private RpcProviderRegistry rpcProviderRegistry = null;
  96     @SuppressWarnings("unused")
  97     private NotificationPublishService notificationPublishService = null;
  98     private ClusterSingletonServiceProvider clusterSingletonServiceProvider;
  99
 100     private final ConcurrentHashMap<String, ONFCoreNetworkElementRepresentation> networkElementRepresentations =
 101             new ConcurrentHashMap<>();
 102
 103     private WebSocketServiceClient webSocketService;
 104     private HtDatabaseEventsService databaseClientEvents;
 105     private ODLEventListener odlEventListener;
 106     private NetconfChangeListener netconfChangeListener;
 107     private DeviceManagerApiServiceImpl rpcApiService;
 108     private @Nullable PerformanceManagerImpl performanceManager = null;
 109     private ProviderClient dcaeProviderClient;
 110     private ProviderClient aotsMProvider;
 111     private @Nullable AaiProviderClient aaiProviderClient;
 112     private DeviceMonitorImpl deviceMonitor;
 113     private IndexUpdateService updateService;
 114     private IndexConfigService configService;
 115     private IndexMwtnService mwtnService;
 116     private HtDatabaseNode htDatabase;
 117     private Boolean devicemanagerInitializationOk = false;
 118     private MaintenanceServiceImpl maintenanceService;
 119     private NotificationDelayService<ProblemNotificationXml> notificationDelayService;
 120     private Thread threadDoClearCurrentFaultByNodename = null;
 121     private int refreshCounter = 0;
 122     private AkkaConfig akkaConfig;
 123     private ArchiveCleanService archiveCleanService;
 124     @SuppressWarnings("unused")
 125         private ClusterSingletonServiceRegistration cssRegistration;
 126
 127     // Blueprint 1
 128     public DeviceManagerImpl() {
 129         LOG.info("Creating provider for {}", APPLICATION_NAME);
 130     }
 131
 132     public void setDataBroker(DataBroker dataBroker) {
 133         this.dataBroker = dataBroker;
 134     }
 135
 136     public void setRpcProviderRegistry(RpcProviderRegistry rpcProviderRegistry) {
 137         this.rpcProviderRegistry = rpcProviderRegistry;
 138     }
 139
 140     public void setNotificationPublishService(NotificationPublishService notificationPublishService) {
 141         this.notificationPublishService = notificationPublishService;
 142     }
 143
 144     public void setMountPointService(MountPointService mountPointService) {
 145         this.mountPointService = mountPointService;
 146     }
 147     public void setClusterSingletonService(ClusterSingletonServiceProvider clusterSingletonService) {
 148         this.clusterSingletonServiceProvider = clusterSingletonService;
 149     }
 150     public void init() {
 151
 152         LOG.info("Session Initiated start {}", APPLICATION_NAME);
 153
 154         // Start RPC Service
 155         this.rpcApiService = new DeviceManagerApiServiceImpl(rpcProviderRegistry);
 156         // Get configuration
 157         HtDevicemanagerConfiguration config = HtDevicemanagerConfiguration.getConfiguration();
 158         try {
 159             this.akkaConfig = AkkaConfig.load();
 160             LOG.debug("akka.conf loaded: " + akkaConfig.toString());
 161         } catch (Exception e1) {
 162             this.akkaConfig = null;
 163             LOG.warn("problem loading akka.conf: " + e1.getMessage());
 164         }
 165         GeoConfig geoConfig = null;
 166         if (akkaConfig != null && akkaConfig.isCluster()) {
 167             LOG.info("cluster mode detected");
 168             if (GeoConfig.fileExists()) {
 169                 try {
 170                     LOG.debug("try to load geoconfig");
 171                     geoConfig = GeoConfig.load();
 172                 } catch (Exception err) {
 173                     LOG.warn("problem loading geoconfig: " + err.getMessage());
 174                 }
 175             } else {
 176                 LOG.debug("no geoconfig file found");
 177             }
 178         } else {
 179             LOG.info("single node mode detected");
 180         }
 181
 182         this.notificationDelayService = new NotificationDelayService<>(config);
 183
 184         EsConfig dbConfig = config.getEs();
 185         LOG.debug("esConfig=" + dbConfig.toString());
 186         // Start database
 187         htDatabase = HtDatabaseNode.start(dbConfig, akkaConfig, geoConfig);
 188         if (htDatabase == null) {
 189             LOG.error("Can only run with local database. Stop initialization of devicemanager.");
 190         } else {
 191             // init Database Values only if singleNode or clusterMember=1
 192             if (akkaConfig == null || akkaConfig.isClusterAndFirstNode()) {
 193                 // Create DB index if not existing and if database is running
 194                 try {
 195                     this.configService = new IndexConfigService(htDatabase);
 196                     this.mwtnService = new IndexMwtnService(htDatabase);
 197                 } catch (Exception e) {
 198                     LOG.warn("Can not start ES access clients to provide database index config, mwtn. ", e);
 199                 }
 200             }
 201             // start service for device maintenance service
 202             this.maintenanceService = new MaintenanceServiceImpl(htDatabase);
 203
 204             // Websockets
 205             try {
 206                 this.webSocketService = new WebSocketServiceClientImpl2(rpcProviderRegistry);
 207             } catch (Exception e) {
 208                 LOG.error("Can not start websocket service. Loading mock class.", e);
 209                 this.webSocketService = new WebSocketServiceClientDummyImpl();
 210             }
 211             // DCAE
 212             this.dcaeProviderClient = new DcaeProviderClient(config, dbConfig.getCluster(), this);
 213
 214             this.aaiProviderClient = new AaiProviderClient(config, this);
 215             // EM
 216             EsConfig emConfig = dbConfig.cloneWithIndex("sdnevents");
 217
 218             if (emConfig == null) {
 219                 LOG.warn("No configuration available. Don't start event manager");
 220             } else {
 221                 this.databaseClientEvents = new HtDatabaseEventsService(htDatabase);
 222                 //Make sure to start for one cluster node only
 223                 if (akkaConfig == null || akkaConfig.isClusterAndFirstNode() || akkaConfig.isSingleNode()) {
 224                  }
 225
 226                 String myDbKeyNameExtended = MYDBKEYNAMEBASE + "-" + dbConfig.getCluster();
 227
 228                 this.odlEventListener = new ODLEventListener(myDbKeyNameExtended, webSocketService,
 229                         databaseClientEvents, dcaeProviderClient, aotsMProvider, maintenanceService);
 230             }
 231             this.archiveCleanService = new ArchiveCleanService(config, databaseClientEvents, mwtnService);
 232             this.cssRegistration = this.clusterSingletonServiceProvider.registerClusterSingletonService(this.archiveCleanService);
 233             // PM
 234             PmConfig configurationPM = config.getPm();
 235             LOG.info("Performance manager configuration: {}", configurationPM);
 236             if (!configurationPM.isPerformanceManagerEnabled()) {
 237
 238                 LOG.info("No configuration available. Don't start performance manager");
 239             } else {
 240                 @Nullable
 241                 MicrowaveHistoricalPerformanceWriterService databaseClientHistoricalPerformance;
 242                 databaseClientHistoricalPerformance = new MicrowaveHistoricalPerformanceWriterService(htDatabase);
 243                 this.performanceManager = new PerformanceManagerImpl(60, databaseClientHistoricalPerformance);
 244             }
 245
 246             // DUS (Database update service)
 247             LOG.debug("start db update service");
 248             this.updateService =
 249                     new IndexUpdateService(htDatabase, dbConfig.getHost(), dbConfig.getCluster(), dbConfig.getNode());
 250             this.updateService.start();
 251
 252             // RPC Service for specific services
 253             this.rpcApiService.setMaintenanceService(this.maintenanceService);
 254             this.rpcApiService.setResyncListener(this);
 255             // DM
 256             // DeviceMonitor has to be available before netconfSubscriptionManager is
 257             // configured
 258             LOG.debug("start DeviceMonitor Service");
 259             this.deviceMonitor = new DeviceMonitorImpl(dataBroker, odlEventListener);
 260
 261             // netconfSubscriptionManager should be the last one because this is a callback
 262             // service
 263             LOG.debug("start NetconfSubscriptionManager Service");
 264             // this.netconfSubscriptionManager = new
 265             // NetconfSubscriptionManagerOfDeviceManager(this, dataBroker);
 266             // this.netconfSubscriptionManager.register();
 267             this.netconfChangeListener = new NetconfChangeListener(this, dataBroker);
 268             this.netconfChangeListener.register();
 269
 270             this.devicemanagerInitializationOk = true;
 271         }
 272         LOG.info("Session Initiated end. Initialization done {}", devicemanagerInitializationOk);
 273     }
 274
 275     @Override
 276     public void close() throws Exception {
 277         LOG.info("DeviceManagerImpl closing ...");
 278
 279         close(performanceManager);
 280         close(dcaeProviderClient);
 281         close(aaiProviderClient);
 282         close(aotsMProvider);
 283         close(deviceMonitor);
 284         close(updateService, configService, mwtnService);
 285         close(htDatabase);
 286         close(netconfChangeListener);
 287         close(maintenanceService);
 288         close(rpcApiService);
 289         close(notificationDelayService);
 290         close(archiveCleanService);
 291         LOG.info("DeviceManagerImpl closing done");
 292     }
 293
 294
 295     /**
 296      * Used to close all Services, that should support AutoCloseable Pattern
 297      *
 298      * @param toClose
 299      * @throws Exception
 300      */
 301     private void close(AutoCloseable... toCloseList) throws Exception {
 302         for (AutoCloseable element : toCloseList) {
 303             if (element != null) {
 304                 element.close();
 305             }
 306         }
 307     }
 308
 309     /*-------------------------------------------------------------------------------------------
 310      * Functions for interface DeviceManagerService
 311      */
 312
 313     /**
 314      * For each mounted device a mountpoint is created and this listener is called.
 315      * Mountpoint was created or existing. Managed device is now fully connected to node/mountpoint.
 316      * @param action provide action
 317      * @param nNodeId id of the mountpoint
 318      * @param nNode mountpoint contents
 319      */
 320     public void startListenerOnNodeForConnectedState(Action action, NodeId nNodeId, NetconfNode nNode) {
 321
 322         String mountPointNodeName = nNodeId.getValue();
 323         LOG.info("Starting Event listener on Netconf for mountpoint {} Action {}", mountPointNodeName, action);
 324
 325         boolean preConditionMissing = false;
 326         if (mountPointService == null) {
 327             preConditionMissing = true;
 328             LOG.warn("No mountservice available.");
 329         }
 330         if (!devicemanagerInitializationOk) {
 331             preConditionMissing = true;
 332             LOG.warn("Devicemanager initialization still pending.");
 333         }
 334         if (preConditionMissing) {
 335             return;
 336         }
 337
 338         if (networkElementRepresentations.containsKey(mountPointNodeName)) {
 339             LOG.warn("Mountpoint {} already registered. Leave startup procedure.", mountPointNodeName);
 340             return;
 341         }
 342
 343         if (!isMaster(nNode)) {
 344             // Change Devicemonitor-status to connected ... for non master mountpoints.
 345             deviceMonitor.deviceConnectSlaveIndication(mountPointNodeName);
 346             return;
 347         }
 348
 349         InstanceIdentifier<Node> instanceIdentifier =
 350                 NETCONF_TOPO_IID.child(Node.class, new NodeKey(new NodeId(mountPointNodeName)));
 351
 352         Optional<MountPoint> optionalMountPoint = null;
 353         int timeout = 10000;
 354         while (!(optionalMountPoint = mountPointService.getMountPoint(instanceIdentifier)).isPresent() && timeout > 0) {
 355             LOG.info("Event listener waiting for mount point for Netconf device :: Name : {}", mountPointNodeName);
 356             sleepMs(1000);
 357             timeout -= 1000;
 358         }
 359
 360         if (!optionalMountPoint.isPresent()) {
 361             LOG.warn("Event listener timeout while waiting for mount point for Netconf device :: Name : {} ",
 362                     mountPointNodeName);
 363             return;
 364         }
 365         // Mountpoint is present for sure
 366         MountPoint mountPoint = optionalMountPoint.get();
 367         //BindingDOMDataBrokerAdapter.BUILDER_FACTORY;
 368         LOG.info("Mountpoint with id: {} class {} toString {}", mountPoint.getIdentifier(), mountPoint.getClass().getName(), mountPoint);
 369         Optional<DataBroker> optionalNetconfNodeDatabroker = mountPoint.getService(DataBroker.class);
 370
 371         if (! optionalNetconfNodeDatabroker.isPresent()) {
 372             LOG.info("Slave mountpoint {} without databroker", mountPointNodeName);
 373             return;
 374         }
 375
 376         DataBroker netconfNodeDataBroker = optionalNetconfNodeDatabroker.get();
 377         LOG.info("Master mountpoint {}", mountPointNodeName);
 378         // getNodeInfoTest(dataBroker);
 379
 380         // create automatic empty maintenance entry into db before reading and listening for problems
 381         this.maintenanceService.createIfNotExists(mountPointNodeName);
 382
 383         // Setup microwaveEventListener for Notificationservice
 384
 385         // MicrowaveEventListener microwaveEventListener = new
 386         // MicrowaveEventListener(mountPointNodeName, websocketmanagerService,
 387         // xmlMapper, databaseClientEvents);
 388         ONFCoreNetworkElementRepresentation ne = ONFCoreNetworkElementFactory.create(mountPointNodeName, dataBroker,
 389                 webSocketService, databaseClientEvents, instanceIdentifier, netconfNodeDataBroker, dcaeProviderClient,
 390                 aotsMProvider, maintenanceService, notificationDelayService);
 391         networkElementRepresentations.put(mountPointNodeName, ne);
 392         ne.doRegisterMicrowaveEventListener(mountPoint);
 393
 394         // Register netconf stream
 395         registerNotificationStream(mountPointNodeName, mountPoint, "NETCONF");
 396
 397         // -- Read data from NE
 398         ne.initialReadFromNetworkElement();
 399         ne.initSynchronizationExtension();
 400
 401         // Setup Service that monitors registration/ deregistration of session
 402         ConnectionStatus csts = nNode.getConnectionStatus();
 403         sendCreateOrUpdateNotification(mountPointNodeName, action, csts);
 404
 405         if (aaiProviderClient != null) {
 406             aaiProviderClient.onDeviceRegistered(mountPointNodeName);
 407         }
 408         // -- Register NE to performance manager
 409         if (performanceManager != null) {
 410             performanceManager.registration(mountPointNodeName, ne);
 411         }
 412
 413         deviceMonitor.deviceConnectMasterIndication(mountPointNodeName, ne);
 414
 415         LOG.info("Starting Event listener on Netconf device :: Name : {} finished", mountPointNodeName);
 416     }
 417
 418     /**
 419      * Mountpoint created or existing. Managed device is actually disconnected from node/ mountpoint.
 420      * Origin state: Connecting, Connected
 421      * Target state: are UnableToConnect or Connecting
 422      * @param action create or update
 423      * @param nNodeId id of the mountpoint
 424      * @param nNode mountpoint contents
 425      */
 426     public void enterNonConnectedState(Action action, NodeId nNodeId, NetconfNode nNode) {
 427         String mountPointNodeName = nNodeId.getValue();
 428         ConnectionStatus csts = nNode.getConnectionStatus();
 429
 430         sendCreateOrUpdateNotification(mountPointNodeName, action, csts);
 431
 432         // Handling if mountpoint exist. connected -> connecting/UnableToConnect
 433         stopListenerOnNodeForConnectedState(mountPointNodeName);
 434
 435         if (deviceMonitor != null) {
 436             deviceMonitor.deviceDisconnectIndication(mountPointNodeName);
 437         }
 438
 439     }
 440
 441     /**
 442      * Mountpoint removed indication.
 443      * @param nNodeId id of the mountpoint
 444      */
 445     public void removeMountpointState(NodeId nNodeId) {
 446         String mountPointNodeName = nNodeId.getValue();
 447         LOG.info("mountpointNodeRemoved {}", nNodeId.getValue());
 448
 449         stopListenerOnNodeForConnectedState(mountPointNodeName);
 450         deviceMonitor.removeMountpointIndication(mountPointNodeName);
 451         if (odlEventListener != null) {
 452             odlEventListener.deRegistration(mountPointNodeName);
 453         }
 454     }
 455
 456     /**
 457      * Do all tasks necessary to move from mountpoint state connected -> connecting
 458      * @param mountPointNodeName provided
 459      * @param ne representing the device connected to mountpoint
 460      */
 461     private void stopListenerOnNodeForConnectedState( String mountPointNodeName) {
 462         ONFCoreNetworkElementRepresentation ne = networkElementRepresentations.remove(mountPointNodeName);
 463         if (ne != null) {
 464             this.maintenanceService.deleteIfNotRequired(mountPointNodeName);
 465             int problems = ne.removeAllCurrentProblemsOfNode();
 466             LOG.debug("Removed all {} problems from database at deregistration for {}", problems, mountPointNodeName);
 467             if (performanceManager != null) {
 468                 performanceManager.deRegistration(mountPointNodeName);
 469             }
 470             if (aaiProviderClient != null) {
 471                 aaiProviderClient.onDeviceUnregistered(mountPointNodeName);
 472             }
 473         }
 474     }
 475
 476     private void sendCreateOrUpdateNotification(String mountPointNodeName, Action action, ConnectionStatus csts) {
 477         LOG.info("enter Non ConnectedState for device :: Name : {} Action {} ConnectionStatus {}", mountPointNodeName, action, csts);
 478         if (action == Action.CREATE) {
 479             odlEventListener.registration(mountPointNodeName);
 480         } else {
 481             odlEventListener.updateRegistration(mountPointNodeName, ConnectionStatus.class.getSimpleName(), csts != null ? csts.getName() : "null");
 482         }
 483     }
 484
 485     /**
 486      * Handle netconf/mountpoint changes
 487      */
 488     @Override
 489     public void netconfChangeHandler(Action action, @Nullable ConnectionStatus csts, NodeId nodeId, NetconfNode nnode) {
 490         switch (action) {
 491             case REMOVE:
 492                 removeMountpointState(nodeId); // Stop Monitor
 493                 //deviceManagerService.enterNonConnectedState(nodeId, nnode); // Remove Mountpoint handler
 494                 break;
 495
 496             case UPDATE:
 497             case CREATE:
 498                 if (csts != null) {
 499                     switch (csts) {
 500                         case Connected: {
 501                             startListenerOnNodeForConnectedState(action, nodeId, nnode);
 502                             break;
 503                         }
 504                         case UnableToConnect:
 505                         case Connecting: {
 506                             enterNonConnectedState(action, nodeId, nnode);
 507                             break;
 508                         }
 509                     }
 510                 } else {
 511                     LOG.debug("NETCONF Node handled with null status for action", action);
 512                 }
 513                 break;
 514         }
 515     }
 516
 517     /*-------------------------------------------------------------------------------------------
 518      * Functions
 519      */
 520
 521     public ArchiveCleanService getArchiveCleanService() {
 522         return this.archiveCleanService;
 523     }
 524
 525     public HtDatabaseEventsService getDatabaseClientEvents() {
 526         return databaseClientEvents;
 527     }
 528
 529     public IndexMwtnService getMwtnService() {
 530         return mwtnService;
 531     }
 532
 533     /**
 534      * Async RPC Interface implementation
 535      */
 536     @Override
 537     public @Nonnull List<String> doClearCurrentFaultByNodename(@Nullable List<String> nodeNamesInput)
 538             throws IllegalStateException {
 539
 540         if (this.databaseClientEvents == null) {
 541             throw new IllegalStateException("dbEvents service not instantiated");
 542         }
 543
 544         if (threadDoClearCurrentFaultByNodename != null && threadDoClearCurrentFaultByNodename.isAlive()) {
 545             throw new IllegalStateException("A clear task is already active");
 546         } else {
 547
 548             // Create list of mountpoints if input is empty, using the content in ES
 549             if (nodeNamesInput == null || nodeNamesInput.size() <= 0) {
 550                 nodeNamesInput = this.databaseClientEvents.getAllNodesWithCurrentAlarms();
 551             }
 552
 553             // Filter all mountpoints from input that were found and are known to this Cluster-node instance of
 554             // DeviceManager
 555             final List<String> nodeNamesHandled = new ArrayList<>();
 556             for (String mountpointName : nodeNamesInput) {
 557                 LOG.info("Work with mountpoint {}", mountpointName);
 558
 559                 if (odlEventListener != null && mountpointName.equals(odlEventListener.getOwnKeyName())) {
 560
 561                     // SDN Controller related alarms
 562                     // -- can not be recreated on all nodes in connected state
 563                     // -- would result in a DCAE/AAI Notification
 564                     // Conclusion for 1810 Delivery ... not covered by RPC function (See issue #43)
 565                     LOG.info("Ignore SDN Controller related alarms for {}", mountpointName);
 566                     // this.databaseClientEvents.clearFaultsCurrentOfNode(mountpointName);
 567                     // nodeNamesHandled.add(mountpointName);
 568
 569                 } else {
 570
 571                     if (mountPointService != null) {
 572                         InstanceIdentifier<Node> instanceIdentifier =
 573                                 NETCONF_TOPO_IID.child(Node.class, new NodeKey(new NodeId(mountpointName)));
 574                         Optional<MountPoint> optionalMountPoint = mountPointService.getMountPoint(instanceIdentifier);
 575
 576                         if (!optionalMountPoint.isPresent()) {
 577                             LOG.info("Remove Alarms for unknown mountpoint {}", mountpointName);
 578                             this.databaseClientEvents.clearFaultsCurrentOfNode(mountpointName);
 579                             nodeNamesHandled.add(mountpointName);
 580                         } else {
 581                             if (networkElementRepresentations.containsKey(mountpointName)) {
 582                                 LOG.info("At node known mountpoint {}", mountpointName);
 583                                 nodeNamesHandled.add(mountpointName);
 584                             } else {
 585                                 LOG.info("At node unknown mountpoint {}", mountpointName);
 586                             }
 587                         }
 588                     }
 589                 }
 590             }
 591
 592             // Force a sync
 593             if (this.deviceMonitor != null) {
 594                 this.deviceMonitor.refreshAlarmsInDb();
 595             }
 596
 597             threadDoClearCurrentFaultByNodename = new Thread(() -> {
 598                 refreshCounter++;
 599                 LOG.info("Start refresh mountpoint task {}", refreshCounter);
 600                 // for(String nodeName:nodeNamesOutput) {
 601                 for (String nodeName : nodeNamesHandled) {
 602                     ONFCoreNetworkElementRepresentation ne = networkElementRepresentations.get(nodeName);
 603                     if (ne != null) {
 604                         LOG.info("Refresh mountpoint {}", nodeName);
 605                         ne.initialReadFromNetworkElement();
 606                     } else {
 607                         LOG.info("Unhandled mountpoint {}", nodeName);
 608                     }
 609                 }
 610                 LOG.info("End refresh mountpoint task {}", refreshCounter);
 611             });
 612             threadDoClearCurrentFaultByNodename.start();
 613             return nodeNamesHandled;
 614         }
 615     };
 616
 617     /**
 618      * Indication if init() of devicemanager successfully done.
 619      *
 620      * @return true if init() was sucessfull. False if not done or not successfull.
 621      */
 622     public boolean isDevicemanagerInitializationOk() {
 623         return this.devicemanagerInitializationOk;
 624     }
 625
 626     /**
 627      * Get initialization status of database.
 628      *
 629      * @return true if fully initialized false if not
 630      */
 631     public boolean isDatabaseInitializationFinished() {
 632         return htDatabase == null ? false : htDatabase.getInitialized();
 633     }
 634
 635     /*---------------------------------------------------------------------
 636      * Private funtions
 637      */
 638
 639     /**
 640      * Do the stream creation for the device.
 641      *
 642      * @param mountPointNodeName
 643      * @param mountPoint
 644      */
 645     private void registerNotificationStream(String mountPointNodeName, MountPoint mountPoint, String streamName) {
 646
 647         final Optional<RpcConsumerRegistry> optionalRpcConsumerService =
 648                 mountPoint.getService(RpcConsumerRegistry.class);
 649         if (optionalRpcConsumerService.isPresent()) {
 650             final RpcConsumerRegistry rpcConsumerRegitry = optionalRpcConsumerService.get();
 651             final NotificationsService rpcService = rpcConsumerRegitry.getRpcService(NotificationsService.class);
 652             if (rpcService == null) {
 653                 LOG.warn("rpcService is null for mountpoint {}", mountPointNodeName);
 654             } else {
 655                 final CreateSubscriptionInputBuilder createSubscriptionInputBuilder =
 656                         new CreateSubscriptionInputBuilder();
 657                 createSubscriptionInputBuilder.setStream(new StreamNameType(streamName));
 658                 LOG.info("Event listener triggering notification stream {} for node {}", streamName,
 659                         mountPointNodeName);
 660                 try {
 661                     CreateSubscriptionInput createSubscriptionInput = createSubscriptionInputBuilder.build();
 662                     if (createSubscriptionInput == null) {
 663                         LOG.warn("createSubscriptionInput is null for mountpoint {}", mountPointNodeName);
 664                     } else {
 665                         rpcService.createSubscription(createSubscriptionInput);
 666                     }
 667                 } catch (NullPointerException e) {
 668                     LOG.warn("createSubscription failed");
 669                 }
 670             }
 671         } else {
 672             LOG.warn("No RpcConsumerRegistry avaialble.");
 673         }
 674
 675     }
 676
 677     /**
 678      * Get NE object
 679      *
 680      * @param mountpoint mount point name
 681      * @return null or NE specific data
 682      */
 683     public @Nullable ONFCoreNetworkElementRepresentation getNeByMountpoint(String mountpoint) {
 684
 685         return networkElementRepresentations.get(mountpoint);
 686
 687     }
 688
 689     /* -- LOG related functions -- */
 690
 691
 692     private boolean isInClusterMode() {
 693         return this.akkaConfig == null ? false : this.akkaConfig.isCluster();
 694     }
 695
 696     private String getClusterNetconfNodeName() {
 697         return this.akkaConfig == null ? "" : this.akkaConfig.getClusterConfig().getClusterSeedNodeName("abc");
 698     }
 699
 700     private boolean isMaster(NetconfNode nnode) {
 701         if (isInClusterMode()) {
 702             LOG.debug("check if me is responsible for node");
 703             String masterNodeName = nnode.getClusteredConnectionStatus() == null ? "null"
 704                     : nnode.getClusteredConnectionStatus().getNetconfMasterNode();
 705             /*
 706              * List<NodeStatus> clusterNodeStatusList=nnode.getClusteredConnectionStatus()==null?null:nnode.
 707              * getClusteredConnectionStatus().getNodeStatus(); if(clusterNodeStatusList!=null) { for(NodeStatus
 708              * s: clusterNodeStatusList) LOG.debug("node "+s.getNode()+
 709              * " with status "+(s.getStatus()==null?"null":s.getStatus().getName())); }
 710              */
 711             String myNodeName = getClusterNetconfNodeName();
 712             LOG.debug("sdnMasterNode=" + masterNodeName + " and sdnMyNode=" + myNodeName);
 713             if (!masterNodeName.equals(myNodeName)) {
 714                 LOG.debug("netconf change but me is not master for this node");
 715                 return false;
 716             }
 717         }
 718         return true;
 719     }
 720
 721
 722     private void sleepMs(int milliseconds) {
 723         try {
 724             Thread.sleep(milliseconds);
 725         } catch (InterruptedException e) {
 726             LOG.debug("Interrupted sleep");
 727             // Restore interrupted state...
 728             Thread.currentThread().interrupt();
 729         }
 730     }
 731
 732 }