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.info("Init task {} handling time {} seconds", PerformanceManagerTask.class.getSimpleName(), seconds);
  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("Scheduler 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      * Task to read PM data from NE
 127      */
 128
 129     /**
 130      * Task runner to read all performance data from Network Elements. Catch exceptions to make sure, that the Task is
 131      * not stopped.
 132      */
 133     @Override
 134     public void run() {
 135
 136         String mountpointName = "No NE";
 137         if (actualNE != null && actualNE.getAcessor().isPresent()) {
 138             mountpointName = actualNE.getAcessor().get().getNodeId().getValue();
 139         }
 140         LOG.debug("{} start {} Start with mountpoint {}", LOGMARKER, tickCounter, mountpointName);
 141
 142         // Proceed to next NE/Interface
 143         getNextInterface(mountpointName);
 144
 145         LOG.debug("{} {} Next interface to handle {}", LOGMARKER, tickCounter,
 146                 actualNE == null ? "No NE/IF" : actualNE.pmStatusToString());
 147
 148         if (actualNE != null) {
 149             try {
 150                 LOG.debug("{} Start to read PM from NE ({})", LOGMARKER, tickCounter);
 151                 Optional<PerformanceDataLtp> allPm = actualNE.getLtpHistoricalPerformanceData();
 152                 if (allPm.isPresent()) {
 153                     LOG.debug("{} {} Got PM list. Start write to DB", LOGMARKER, tickCounter);
 154                     databaseService.doWritePerformanceData(allPm.get().getList());
 155                 }
 156                 LOG.debug("{} {} PM List end.", LOGMARKER, tickCounter);
 157             } catch (Throwable e) {
 158                 LOG.debug("{} {} PM Exception", LOGMARKER, tickCounter);
 159                 String msg = new StringBuffer().append(e.getMessage()).toString();
 160                 LOG.warn("{} {} PM read/write failed. Write log entry {}", LOGMARKER, tickCounter, msg);
 161                 netconfNetworkElementService.writeToEventLog(mountpointName, "PM Problem", msg);
 162             }
 163         }
 164
 165         LOG.debug("{} end {}", LOGMARKER, tickCounter);
 166         tickCounter++;
 167     }
 168
 169     /**
 170      * Reset queue to start from beginning
 171      */
 172     private void resetQueue() {
 173         actualNE = null;
 174         neIterator = null;
 175     }
 176
 177     /**
 178      * Get then next interface in the list. First try to find a next on the actual NE. If not available search next
 179      * interface at a NE Special Situations to handle: Empty queue, NEs, but no interfaces
 180      */
 181     private void getNextInterface(String mountpointName) {
 182         boolean started = false;
 183         int loopCounter = 0;
 184
 185         LOG.debug("{} {} getNextInterface enter. Queue size {} ", LOGMARKER, tickCounter, queue.size());
 186
 187         if (actualNE != null && !queue.containsValue(actualNE)) {
 188             LOG.debug("{} {} NE Removed duringprocessing A", LOGMARKER, tickCounter);
 189             resetQueue();
 190         }
 191
 192         while (true) {
 193
 194             if (loopCounter++ >= 1000) {
 195                 LOG.error("{} {} Problem in PM iteration. endless condition reached", LOGMARKER, tickCounter);
 196                 resetQueue();
 197                 break;
 198             }
 199
 200             LOG.debug("{} {} Loop ne {}:neiterator {}:Interfaceiterator:{} Loop:{}", LOGMARKER, tickCounter,
 201                     actualNE == null ? "null" : mountpointName, neIterator == null ? "null" : neIterator.hasNext(),
 202                     actualNE == null ? "null" : actualNE.hasNext(), loopCounter);
 203
 204             if (actualNE != null && actualNE.hasNext()) {
 205                 // Yes, there is an interface, deliver back
 206                 LOG.debug("{} {} getNextInterface yes A", LOGMARKER, tickCounter);
 207                 actualNE.next();
 208                 break;
 209
 210             } else {
 211                 // No element in neInterfaceInterator .. get next NE and try
 212                 if (neIterator != null && neIterator.hasNext()) {
 213                     // Set a new NE
 214                     LOG.debug("{} {} Next NE A", LOGMARKER, tickCounter);
 215                     actualNE = neIterator.next();
 216                     actualNE.resetPMIterator();
 217
 218                 } else {
 219                     // Goto start condition 1) first entry 2) end of queue reached
 220                     LOG.debug("{} {} Reset", LOGMARKER, tickCounter);
 221                     resetQueue();
 222
 223                     if (queue.isEmpty()) {
 224                         LOG.debug("{} {} no nextInterfac. queue empty", LOGMARKER, tickCounter);
 225                         break;
 226                     } else if (!started) {
 227                         LOG.debug("{} {} getNextInterface start condition. Get interator.", LOGMARKER, tickCounter);
 228                         neIterator = queue.values().iterator();
 229                         started = true;
 230                     } else {
 231                         LOG.debug("{} {} no nextInterface", LOGMARKER, tickCounter);
 232                         break;
 233                     }
 234                 }
 235             }
 236         } // while
 237
 238         if (actualNE != null && !queue.containsValue(actualNE)) {
 239             LOG.debug("{} {} NE Removed duringprocessing B", LOGMARKER, tickCounter);
 240             resetQueue();
 241         }
 242
 243     }
 244 }