gerrit.onap Code Review - ccsdk/features.git/blob

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