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

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