sdnr/wt/devicemanager-core/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/devicemanager/performancemanager/impl/PerformanceManagerTask.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.performancemanager.impl;
  19
  20 import java.util.Iterator;
  21 import java.util.Optional;
  22 import java.util.concurrent.ConcurrentHashMap;
  23 import java.util.concurrent.Executors;
  24 import java.util.concurrent.ScheduledExecutorService;
  25 import java.util.concurrent.ScheduledFuture;
  26 import java.util.concurrent.TimeUnit;
  27
  28 import org.onap.ccsdk.features.sdnr.wt.dataprovider.model.DataProvider;
  29 import org.onap.ccsdk.features.sdnr.wt.devicemanager.ne.service.NetworkElement;
  30 import org.onap.ccsdk.features.sdnr.wt.devicemanager.ne.service.PerformanceDataProvider;
  31 import org.onap.ccsdk.features.sdnr.wt.devicemanager.service.NetconfNetworkElementService;
  32 import org.onap.ccsdk.features.sdnr.wt.devicemanager.types.PerformanceDataLtp;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 public class PerformanceManagerTask implements Runnable {
  37
  38         private static final Logger LOG = LoggerFactory.getLogger(PerformanceManagerTask.class);
  39         private static final String LOGMARKER = "PMTick";
  40
  41         private int tickCounter = 0;
  42
  43         private final ConcurrentHashMap<String, PerformanceDataProvider> queue = new ConcurrentHashMap<>();
  44         private final DataProvider databaseService;
  45         private final ScheduledExecutorService scheduler;
  46         private final long seconds;
  47
  48         private ScheduledFuture<?> taskHandle = null;
  49         private Iterator<PerformanceDataProvider> neIterator = null;
  50         private PerformanceDataProvider actualNE = null;
  51         private final NetconfNetworkElementService netconfNetworkElementService;
  52
  53         /**
  54          * Constructor of PM Task
  55          *
  56          * @param seconds                                     seconds to call PM Task
  57          * @param microwaveHistoricalPerformanceWriterService DB Service to load PM data
  58          *                                                    to
  59          * @param netconfNetworkElementService                to write into log
  60          */
  61
  62         public PerformanceManagerTask(long seconds, DataProvider microwaveHistoricalPerformanceWriterService,
  63                         NetconfNetworkElementService netconfNetworkElementService) {
  64
  65                 LOG.info("Init task {} handling time {} seconds", PerformanceManagerTask.class.getSimpleName(), seconds);
  66                 this.seconds = seconds;
  67                 this.databaseService = microwaveHistoricalPerformanceWriterService;
  68                 this.scheduler = Executors.newSingleThreadScheduledExecutor();
  69                 this.netconfNetworkElementService = netconfNetworkElementService;
  70
  71         }
  72
  73         /**
  74          * Start PM Task
  75          */
  76         public void start() {
  77                 LOG.info("PM task created");
  78                 taskHandle = this.scheduler.scheduleAtFixedRate(this, 0, seconds, TimeUnit.SECONDS);
  79                 LOG.info("PM task scheduled");
  80         }
  81
  82         /**
  83          * Stop everything
  84          */
  85         public void stop() {
  86                 LOG.info("Stop {}", PerformanceManagerImpl.class.getSimpleName());
  87                 if (taskHandle != null) {
  88                         taskHandle.cancel(true);
  89                         try {
  90                                 scheduler.awaitTermination(10, TimeUnit.SECONDS);
  91                         } catch (InterruptedException e) {
  92                                 LOG.debug("Scheduler stopped.", e);
  93                                 // Restore interrupted state...
  94                                 Thread.currentThread().interrupt();
  95                         }
  96                 }
  97         }
  98
  99         /**
 100          * Add NE/Mountpoint to PM Processig
 101          *
 102          * @param mountPointNodeName to be added
 103          * @param ne                 that is connected to the mountpoint
 104          */
 105         public void registration(String mountPointNodeName, NetworkElement ne) {
 106
 107                 Optional<PerformanceDataProvider> oPmNe = ne.getService(PerformanceDataProvider.class);
 108                 if (oPmNe.isPresent()) {
 109                         queue.put(mountPointNodeName, oPmNe.get());
 110                 }
 111         }
 112
 113         /**
 114          * Remove mountpoint/NE from PM process
 115          *
 116          * @param mountPointNodeName that has to be removed
 117          */
 118         public void deRegistration(String mountPointNodeName) {
 119                 LOG.debug("Deregister {}", mountPointNodeName);
 120                 PerformanceDataProvider removedNE = queue.remove(mountPointNodeName);
 121
 122                 if (removedNE == null) {
 123                         LOG.warn("Couldn't delete {}", mountPointNodeName);
 124                 }
 125         }
 126
 127         /*--------------------------------------------------------------
 128          * Task to read PM data from NE
 129          */
 130
 131         /**
 132          * Task runner to read all performance data from Network Elements. Catch
 133          * exceptions to make sure, that the Task is not stopped.
 134          */
 135         @Override
 136         public void run() {
 137
 138                 String mountpointName = "No NE";
 139                 if (actualNE != null && actualNE.getAcessor().isPresent()) {
 140                         mountpointName = actualNE.getAcessor().get().getNodeId().getValue();
 141                 }
 142                 LOG.debug("{} start {} Start with mountpoint {}", LOGMARKER, tickCounter, mountpointName);
 143
 144                 // Proceed to next NE/Interface
 145                 getNextInterface(mountpointName);
 146
 147                 LOG.debug("{} {} Next interface to handle {}", LOGMARKER, tickCounter,
 148                                 actualNE == null ? "No NE/IF" : actualNE.pmStatusToString());
 149
 150                 if (actualNE != null) {
 151                         try {
 152                                 LOG.debug("{} Start to read PM from NE ({})", LOGMARKER, tickCounter);
 153                                 Optional<PerformanceDataLtp> allPm = actualNE.getLtpHistoricalPerformanceData();
 154                                 if (allPm.isPresent()) {
 155                                         LOG.debug("{} {} Got PM list. Start write to DB", LOGMARKER, tickCounter);
 156                                         databaseService.doWritePerformanceData(allPm.get().getList());
 157                                 }
 158                                 LOG.debug("{} {} PM List end.", LOGMARKER, tickCounter);
 159                         } catch (Throwable e) {
 160                                 LOG.debug("{} {} PM Exception", LOGMARKER, tickCounter);
 161                                 String msg = new StringBuffer().append(e.getMessage()).toString();
 162                                 LOG.warn("{} {} PM read/write failed. Write log entry {}", LOGMARKER, tickCounter, msg);
 163                                 netconfNetworkElementService.writeToEventLog(mountpointName, "PM Problem", msg);
 164                         }
 165                 }
 166
 167                 LOG.debug("{} end {}", LOGMARKER, tickCounter);
 168                 tickCounter++;
 169         }
 170
 171         /**
 172          * Reset queue to start from beginning
 173          */
 174         private void resetQueue() {
 175                 actualNE = null;
 176                 neIterator = null;
 177         }
 178
 179         /**
 180          * Get then next interface in the list. First try to find a next on the actual
 181          * NE. If not available search next interface at a NE Special Situations to
 182          * handle: Empty queue, NEs, but no interfaces
 183          */
 184         private void getNextInterface(String mountpointName) {
 185                 boolean started = false;
 186                 int loopCounter = 0;
 187
 188                 LOG.debug("{} {} getNextInterface enter. Queue size {} ", LOGMARKER, tickCounter, queue.size());
 189
 190                 if (actualNE != null && !queue.containsValue(actualNE)) {
 191                         LOG.debug("{} {} NE Removed duringprocessing A", LOGMARKER, tickCounter);
 192                         resetQueue();
 193                 }
 194
 195                 while (true) {
 196
 197                         if (loopCounter++ >= 1000) {
 198                                 LOG.error("{} {} Problem in PM iteration. endless condition reached", LOGMARKER, tickCounter);
 199                                 resetQueue();
 200                                 break;
 201                         }
 202
 203                         LOG.debug("{} {} Loop ne {}:neiterator {}:Interfaceiterator:{} Loop:{}", LOGMARKER, tickCounter,
 204                                         actualNE == null ? "null" : mountpointName, neIterator == null ? "null" : neIterator.hasNext(),
 205                                         actualNE == null ? "null" : actualNE.hasNext(), loopCounter);
 206
 207                         if (actualNE != null && actualNE.hasNext()) {
 208                                 // Yes, there is an interface, deliver back
 209                                 LOG.debug("{} {} getNextInterface yes A", LOGMARKER, tickCounter);
 210                                 actualNE.next();
 211                                 break;
 212
 213                         } else {
 214                                 // No element in neInterfaceInterator .. get next NE and try
 215                                 if (neIterator != null && neIterator.hasNext()) {
 216                                         // Set a new NE
 217                                         LOG.debug("{} {} Next NE A", LOGMARKER, tickCounter);
 218                                         actualNE = neIterator.next();
 219                                         actualNE.resetPMIterator();
 220
 221                                 } else {
 222                                         // Goto start condition 1) first entry 2) end of queue reached
 223                                         LOG.debug("{} {} Reset", LOGMARKER, tickCounter);
 224                                         resetQueue();
 225
 226                                         if (queue.isEmpty()) {
 227                                                 LOG.debug("{} {} no nextInterfac. queue empty", LOGMARKER, tickCounter);
 228                                                 break;
 229                                         } else if (!started) {
 230                                                 LOG.debug("{} {} getNextInterface start condition. Get interator.", LOGMARKER, tickCounter);
 231                                                 neIterator = queue.values().iterator();
 232                                                 started = true;
 233                                         } else {
 234                                                 LOG.debug("{} {} no nextInterface", LOGMARKER, tickCounter);
 235                                                 break;
 236                                         }
 237                                 }
 238                         }
 239                 } // while
 240
 241                 if (actualNE != null && !queue.containsValue(actualNE)) {
 242                         LOG.debug("{} {} NE Removed duringprocessing B", LOGMARKER, tickCounter);
 243                         resetQueue();
 244                 }
 245
 246         }
 247 }