sdnr/wireless-transport/code-Carbon-SR1/apps/devicemanager/impl/src/main/java/org/opendaylight/mwtn/devicemanager/impl/DeviceManagerImpl.java

   1 /**
   2  * Copyright (c) 2017 highstreet technologies GmbH
   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
   9 package org.opendaylight.mwtn.devicemanager.impl;
  10
  11 import java.util.concurrent.ConcurrentHashMap;
  12
  13 import javax.annotation.Nullable;
  14
  15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  16 import org.opendaylight.controller.md.sal.binding.api.MountPoint;
  17 import org.opendaylight.controller.md.sal.binding.api.MountPointService;
  18 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ProviderContext;
  19 import org.opendaylight.controller.sal.binding.api.BindingAwareProvider;
  20 import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
  21 import org.opendaylight.mwtn.aaiConnector.impl.AaiProviderClient;
  22 import org.opendaylight.mwtn.base.database.HtDatabaseNode;
  23 import org.opendaylight.mwtn.base.netconf.ONFCoreNetworkElementFactory;
  24 import org.opendaylight.mwtn.base.netconf.ONFCoreNetworkElementRepresentation;
  25 import org.opendaylight.mwtn.config.impl.AkkaConfig;
  26 import org.opendaylight.mwtn.config.impl.EsConfig;
  27 import org.opendaylight.mwtn.config.impl.GeoConfig;
  28 import org.opendaylight.mwtn.config.impl.HtDevicemanagerConfiguration;
  29 import org.opendaylight.mwtn.config.impl.PmConfig;
  30 import org.opendaylight.mwtn.dcaeConnector.impl.DcaeProviderClient;
  31 import org.opendaylight.mwtn.deviceMonitor.impl.DeviceMonitorImpl;
  32 import org.opendaylight.mwtn.devicemanager.api.DeviceManagerService;
  33 import org.opendaylight.mwtn.devicemanager.impl.database.service.HtDatabaseEventsService;
  34 import org.opendaylight.mwtn.devicemanager.impl.listener.NetconfChangeListener;
  35 import org.opendaylight.mwtn.devicemanager.impl.listener.ODLEventListener;
  36 import org.opendaylight.mwtn.devicemanager.impl.xml.WebSocketServiceClient;
  37 import org.opendaylight.mwtn.devicemanager.impl.xml.XmlMapper;
  38 import org.opendaylight.mwtn.index.impl.IndexConfigService;
  39 import org.opendaylight.mwtn.index.impl.IndexMwtnService;
  40 import org.opendaylight.mwtn.index.impl.IndexUpdateService;
  41 import org.opendaylight.mwtn.performancemanager.impl.PerformanceManagerImpl;
  42 import org.opendaylight.mwtn.performancemanager.impl.database.service.MicrowaveHistoricalPerformanceWriterService;
  43 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
  44 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInputBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.NotificationsService;
  46 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.network.topology.topology.topology.types.TopologyNetconf;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.websocketmanager.rev150105.WebsocketmanagerService;
  50 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  51 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  52 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  53 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  54 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  55 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  56 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  57 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  58 import org.slf4j.Logger;
  59 import org.slf4j.LoggerFactory;
  60
  61 import com.google.common.base.Optional;
  62
  63 public class DeviceManagerImpl implements DeviceManagerService, BindingAwareProvider, AutoCloseable {
  64
  65         private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
  66         private static final String MYDBKEYNAME = "SDN-Controller";
  67
  68         // http://sendateodl:8181/restconf/operational/network-topology:network-topology/topology/topology-netconf
  69         private static final InstanceIdentifier<Topology> NETCONF_TOPO_IID = InstanceIdentifier
  70                         .create(NetworkTopology.class)
  71                         .child(Topology.class, new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())));
  72
  73         private ProviderContext session;
  74         private DataBroker dataBroker;
  75
  76         private WebSocketServiceClient webSocketService;
  77         private HtDatabaseEventsService databaseClientEvents;
  78         private ODLEventListener odlEventListener;
  79         private NetconfChangeListener netconfChangeListener;
  80
  81         private final ConcurrentHashMap<String, ONFCoreNetworkElementRepresentation> networkElementRepresentations = new ConcurrentHashMap<>();
  82         private @Nullable PerformanceManagerImpl performanceManager = null;
  83         private ProviderClient dcaeProviderClient;
  84         private @Nullable AaiProviderClient aaiProviderClient;
  85         private DeviceMonitorImpl deviceMonitor;
  86         private IndexUpdateService updateService;
  87         private IndexConfigService configService;
  88         private IndexMwtnService mwtnService;
  89         private HtDatabaseNode htDatabase;
  90         private Boolean initialized = false;
  91
  92         @Override
  93         public void onSessionInitiated(ProviderContext pSession) {
  94                 LOG.info("Session Initiated start");
  95
  96                 this.session = pSession;
  97                 this.dataBroker = pSession.getSALService(DataBroker.class);
  98
  99                 // Get configuration
 100                 HtDevicemanagerConfiguration config = HtDevicemanagerConfiguration.getConfiguration();
 101                 AkkaConfig akkaConfig = null;
 102                 try {
 103                         akkaConfig = AkkaConfig.load();
 104                 } catch (Exception e1) {
 105                         LOG.warn("error loading akka.conf: " + e1.getMessage());
 106                 }
 107                 GeoConfig geoConfig = null;
 108                 if (akkaConfig != null && akkaConfig.isCluster()) {
 109                         if (GeoConfig.fileExists()) {
 110                                 try {
 111                                         LOG.debug("try to load geoconfig");
 112                                         geoConfig = GeoConfig.load();
 113                                 } catch (Exception err) {
 114                                         LOG.warn("error loading geoconfig: " + err.getMessage());
 115                                 }
 116                         } else
 117                                 LOG.debug("no geoconfig file found");
 118                 }
 119                 EsConfig dbConfig = config.getEs();
 120                 LOG.debug("esConfig=" + dbConfig.toString());
 121                 // Start database
 122                 htDatabase = HtDatabaseNode.start(dbConfig, akkaConfig,geoConfig);
 123
 124                 // init Database Values only if singleNode or clusterMember=1
 125                 if (akkaConfig == null || akkaConfig.isSingleNode() || (akkaConfig != null && akkaConfig.isCluster()
 126                                 && akkaConfig.getClusterConfig().getRoleMemberIndex() == 1)) {
 127                         // Create DB index if not existing and if database is running
 128                         this.configService = new IndexConfigService(htDatabase);
 129                         this.mwtnService = new IndexMwtnService(htDatabase);
 130                 }
 131                 // Websockets
 132                 this.webSocketService = new WebSocketServiceClient(pSession.getRpcService(WebsocketmanagerService.class),
 133                                 new XmlMapper());
 134
 135                 // DCAE
 136                 this.dcaeProviderClient = new DcaeProviderClient(config, dbConfig.getCluster(), this);
 137
 138                 this.aaiProviderClient = new AaiProviderClient(config,this);
 139                 // EM
 140                 EsConfig emConfig = dbConfig.cloneWithIndex("sdnevents");
 141
 142                 if (emConfig == null) {
 143                         LOG.warn("No {} configuration available. Don't start event manager");
 144                 } else {
 145                         this.databaseClientEvents = new HtDatabaseEventsService(htDatabase);
 146
 147                         String myDbKeyNameExtended=MYDBKEYNAME+"-"+dbConfig.getCluster();
 148
 149
 150                         this.odlEventListener = new ODLEventListener(myDbKeyNameExtended, webSocketService, databaseClientEvents,
 151                                         dcaeProviderClient, null);
 152                 }
 153
 154                 // PM
 155                 PmConfig configurationPM = config.getPm();
 156                 LOG.info("Performance manager configuration: {}", configurationPM);
 157                 if (!configurationPM.isPerformanceManagerEnabled()) {
 158
 159                         LOG.info("No configuration available. Don't start performance manager");
 160                 } else {
 161                         @Nullable MicrowaveHistoricalPerformanceWriterService databaseClientHistoricalPerformance;
 162                         databaseClientHistoricalPerformance = new MicrowaveHistoricalPerformanceWriterService(htDatabase);
 163                         this.performanceManager = new PerformanceManagerImpl(60, databaseClientHistoricalPerformance);
 164                 }
 165
 166                 // DUS (Database update service)
 167                 LOG.debug("start db update service");
 168                 this.updateService = new IndexUpdateService(htDatabase, dbConfig.getHost(), dbConfig.getCluster(),
 169                                 dbConfig.getNode());
 170                 this.updateService.start();
 171
 172                 // DM
 173                 // DeviceMonitor has to be available before netconfSubscriptionManager is
 174                 // configured
 175                 LOG.debug("start DeviceMonitor Service");
 176                 this.deviceMonitor = new DeviceMonitorImpl(odlEventListener);
 177
 178                 // netconfSubscriptionManager should be the last one because this is a callback
 179                 // service
 180                 LOG.debug("start NetconfSubscriptionManager Service");
 181                 // this.netconfSubscriptionManager = new
 182                 // NetconfSubscriptionManagerOfDeviceManager(this, dataBroker);
 183                 // this.netconfSubscriptionManager.register();
 184                 this.netconfChangeListener = new NetconfChangeListener(this, dataBroker);
 185                 this.netconfChangeListener.register();
 186
 187                 synchronized (initialized) {
 188                         initialized = true;
 189                 }
 190
 191                 LOG.info("Session Initiated end");
 192         }
 193
 194         @Override
 195         public void close() throws Exception {
 196                 LOG.info("DeviceManagerImpl closing ...");
 197
 198                 close(performanceManager);
 199                 close(dcaeProviderClient);
 200                 close(aaiProviderClient);
 201                 close(deviceMonitor);
 202                 close(updateService, configService, mwtnService);
 203                 close(htDatabase);
 204                 close(netconfChangeListener);
 205
 206                 LOG.info("DeviceManagerImpl closing done");
 207         }
 208
 209         /**
 210          * Used to close all Services, that should support AutoCloseable Pattern
 211          *
 212          * @param toClose
 213          * @throws Exception
 214          */
 215         private void close(AutoCloseable... toCloseList) throws Exception {
 216                 for (int t = 0; t < toCloseList.length; t++) {
 217                         if (toCloseList[t] != null)
 218                                 toCloseList[t].close();
 219                 }
 220         }
 221
 222         /**
 223          * For each mounted device a mountpoint is created and this listener is called.
 224          */
 225         @Override
 226         public void startListenerOnNode(NodeId nNodeId, NetconfNode nNode) {
 227                 synchronized (networkElementRepresentations) {
 228
 229                         String mountPointNodeName = nNodeId.getValue();
 230                         LOG.info("Starting Event listener on Netconf device :: Name : {}", mountPointNodeName);
 231
 232                         if (networkElementRepresentations.containsKey(mountPointNodeName)) {
 233                                 LOG.warn("Mountpoint {} already registered. Leave startup procedure.", mountPointNodeName);
 234                                 return;
 235                         }
 236                         if (!initialized) {
 237                                 LOG.warn("Devicemanager initialization still pending. Leave startup procedure. Mountpoint {}", mountPointNodeName);
 238                                 return;
 239                         }
 240
 241                         MountPointService mountService = session.getSALService(MountPointService.class);
 242
 243                         InstanceIdentifier<Node> instanceIdentifier = NETCONF_TOPO_IID.child(Node.class,
 244                                         new NodeKey(new NodeId(mountPointNodeName)));
 245
 246                         Optional<MountPoint> optionalMountPoint = null;
 247                         int timeout = 10000;
 248                         while (!(optionalMountPoint = mountService.getMountPoint(instanceIdentifier)).isPresent() && timeout > 0) {
 249
 250                                 LOG.info("Event listener waiting for mount point for Netconf device :: Name : {}", mountPointNodeName);
 251                                 try {
 252                                         Thread.sleep(1000);
 253                                         timeout -= 1000;
 254                                 } catch (InterruptedException e) {
 255                                         LOG.info("Event listener waiting for mount point for Netconf device :: Name : {} Time: {}",
 256                                                         mountPointNodeName, timeout);
 257                                 }
 258                         }
 259
 260                         if (!optionalMountPoint.isPresent()) {
 261                                 LOG.warn("Event listener timeout while waiting for mount point for Netconf device :: Name : {} ",
 262                                                 mountPointNodeName);
 263                                 return;
 264                         }
 265
 266                         // Mountpoint is present for sure
 267                         MountPoint mountPoint = optionalMountPoint.get();
 268
 269                         DataBroker netconfNodeDataBroker = mountPoint.getService(DataBroker.class).orNull();
 270                         if (netconfNodeDataBroker == null) {
 271                                 LOG.info("Mountpoint is slave mountpoint {}", mountPointNodeName);
 272                                 return;
 273                         }
 274
 275                         LOG.info("Databroker service 1:{} 2:{}", dataBroker.hashCode(), netconfNodeDataBroker.hashCode());
 276                         // getNodeInfoTest(dataBroker);
 277
 278
 279                         // Setup microwaveEventListener for Notificationservice
 280
 281                         // MicrowaveEventListener microwaveEventListener = new
 282                         // MicrowaveEventListener(mountPointNodeName, websocketmanagerService,
 283                         // xmlMapper, databaseClientEvents);
 284                         ONFCoreNetworkElementRepresentation ne = ONFCoreNetworkElementFactory.create(mountPointNodeName, dataBroker,
 285                                         webSocketService, databaseClientEvents, instanceIdentifier, netconfNodeDataBroker, dcaeProviderClient,
 286                                         null);
 287                         networkElementRepresentations.put(mountPointNodeName, ne);
 288                         ne.doRegisterMicrowaveEventListener(mountPoint);
 289
 290                         // Register netconf stream
 291                         registerNotificationStream(mountPointNodeName, mountPoint, "NETCONF");
 292
 293                         // -- Read data from NE
 294                         ne.initialReadFromNetworkElement();
 295                         ne.initSynchronizationExtension();
 296
 297                         // Setup Service that monitors registration/ deregistration of session
 298                         odlEventListener.registration(mountPointNodeName);
 299
 300                         if (aaiProviderClient != null) {
 301                                 aaiProviderClient.onDeviceRegistered(mountPointNodeName);
 302                         }
 303                         // -- Register NE to performance manager
 304                         if (performanceManager != null) {
 305                                 performanceManager.registration(mountPointNodeName, ne);
 306                         }
 307
 308                         deviceMonitor.deviceConnectIndication(mountPointNodeName, ne);
 309
 310                         LOG.info("Starting Event listener on Netconf device :: Name : {} finished", mountPointNodeName);
 311
 312                 }
 313
 314         }
 315
 316         @Override
 317         public void removeListenerOnNode(NodeId nNodeId, NetconfNode nNode) {
 318                 String mountPointNodeName = nNodeId.getValue();
 319                 LOG.info("Removing NetworkElementRepresetations for device :: Name : {}", mountPointNodeName);
 320
 321                 ONFCoreNetworkElementRepresentation ne = networkElementRepresentations.remove(mountPointNodeName);
 322                 if (ne != null) {
 323                         int problems = ne.removeAllCurrentProblemsOfNode();
 324                         LOG.debug("Removed all {} problems from database at deregistration for {}", problems, mountPointNodeName);
 325                         if (odlEventListener != null)
 326                                 odlEventListener.deRegistration(mountPointNodeName);
 327                         if (performanceManager != null) {
 328                                 performanceManager.deRegistration(mountPointNodeName);
 329                         }
 330                         if (aaiProviderClient != null) {
 331                                 aaiProviderClient.onDeviceUnregistered(mountPointNodeName);
 332                         }
 333                 } else {
 334                         LOG.info("No related ne object for mountpoint {} to deregister .", mountPointNodeName);
 335                 }
 336                 if (deviceMonitor != null)
 337                         deviceMonitor.deviceDisconnectIndication(mountPointNodeName);
 338
 339         }
 340
 341         @Override
 342         public void mountpointNodeCreation(NodeId nNodeId, NetconfNode nNode) {
 343                 String mountPointNodeName = nNodeId.getValue();
 344                 LOG.info("mountpointNodeCreation {} {}", nNodeId.getValue(), nNode.getConnectionStatus());
 345                 deviceMonitor.createMountpointIndication(mountPointNodeName);
 346         }
 347
 348         @Override
 349         public void mountpointNodeRemoved(NodeId nNodeId) {
 350                 String mountPointNodeName = nNodeId.getValue();
 351                 LOG.info("mountpointNodeRemoved {}", nNodeId.getValue());
 352                 deviceMonitor.removeMountpointIndication(mountPointNodeName);
 353         }
 354
 355         /**
 356          * Do the stream creation for the device.
 357          *
 358          * @param mountPointNodeName
 359          * @param mountPoint
 360          */
 361         private void registerNotificationStream(String mountPointNodeName, MountPoint mountPoint, String streamName) {
 362
 363                 final Optional<RpcConsumerRegistry> optionalRpcConsumerService = mountPoint
 364                                 .getService(RpcConsumerRegistry.class);
 365                 if (optionalRpcConsumerService.isPresent()) {
 366                         final RpcConsumerRegistry rpcConsumerRegitry = optionalRpcConsumerService.get();
 367                         final NotificationsService rpcService = rpcConsumerRegitry.getRpcService(NotificationsService.class);
 368                         if (rpcService == null) {
 369                                 LOG.warn("rpcService is null for mountpoint {}", mountPointNodeName);
 370                         } else {
 371                                 final CreateSubscriptionInputBuilder createSubscriptionInputBuilder = new CreateSubscriptionInputBuilder();
 372                                 createSubscriptionInputBuilder.setStream(new StreamNameType(streamName));
 373                                 LOG.info("Event listener triggering notification stream {} for node {}", streamName, mountPointNodeName);
 374                                 try {
 375                                         CreateSubscriptionInput createSubscriptionInput = createSubscriptionInputBuilder.build();
 376                                         if (createSubscriptionInput == null) {
 377                                                 LOG.warn("createSubscriptionInput is null for mountpoint {}", mountPointNodeName);
 378                                         } else {
 379                                             rpcService.createSubscription(createSubscriptionInput);
 380                                         }
 381                                 } catch (NullPointerException e) {
 382                                         LOG.warn("createSubscription failed");
 383                                 }
 384                         }
 385                 } else {
 386                         LOG.warn("No RpcConsumerRegistry avaialble.");
 387                 }
 388
 389         }
 390
 391         /**
 392          * Get NE object
 393          * @param mountpoint
 394          * @return null or ne specific data
 395          */
 396         public @Nullable ONFCoreNetworkElementRepresentation getNeByMountpoint( String mountpoint ) {
 397
 398                 return networkElementRepresentations.get(mountpoint);
 399
 400         }
 401
 402 }