dblib/provider/src/main/java/org/onap/ccsdk/sli/core/dblib/pm/PollingWorker.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * onap
   4  * ================================================================================
   5  * Copyright (C) 2016 - 2017 ONAP
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.ccsdk.sli.core.dblib.pm;
  22
  23 import java.util.Iterator;
  24 import java.util.Properties;
  25 import java.util.Set;
  26 import java.util.Timer;
  27 import java.util.TimerTask;
  28 import java.util.TreeSet;
  29
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 import java.util.concurrent.LinkedBlockingQueue;
  34 import java.util.concurrent.atomic.AtomicLong;
  35
  36 public class PollingWorker implements Runnable {
  37
  38         private Logger LOGGER = LoggerFactory.getLogger(PollingWorker.class);
  39
  40         private static PollingWorker self = null;
  41
  42         private LinkedBlockingQueue tasks = new LinkedBlockingQueue(100);
  43         private long interval = 1000L;
  44         private Thread worker = null;
  45         private AtomicLong[] counters = null;
  46         private int[] bucketUnit = null;
  47         private static boolean enabled = false;
  48         private Timer timer = null;
  49
  50         public static void post(long starttime){
  51                 PollingWorker temp = self;
  52                 if(temp != null && enabled) {
  53                         temp.register(new TestSample(starttime));
  54                 }
  55         }
  56
  57         public static void createInistance(Properties props){
  58                 self = new PollingWorker(props);
  59         }
  60
  61         private PollingWorker(Properties ctxprops){
  62                 if(ctxprops==null ||  ctxprops.getProperty("org.onap.ccsdk.dblib.pm") == null){
  63                         enabled = false;
  64                 } else {
  65                         if("true".equalsIgnoreCase((String)ctxprops.getProperty("org.onap.ccsdk.dblib.pm"))){
  66                                 enabled = true;
  67                         } else {
  68                                 enabled = false;
  69                         }
  70                 }
  71
  72                 interval = Long.parseLong(( ctxprops == null || ctxprops.getProperty("org.onap.ccsdk.dblib.pm.interval") == null) ? "60" : (String)ctxprops.getProperty("org.onap.ccsdk.dblib.pm.interval"));
  73                 // '0' bucket is to count exceptions
  74                 String sampling[] = ((ctxprops == null || ctxprops.getProperty("org.onap.ccsdk.dblib.pm.sampling")==null) ? "0,2,5,10,20,50,100" :      (String)ctxprops.getProperty("org.onap.ccsdk.dblib.pm.sampling")).split(",");
  75
  76                 if(enabled){
  77                         bucketUnit = new int[sampling.length];
  78                         for(int i=0, max = bucketUnit.length; i<max; i++){
  79                                 bucketUnit[i] = Integer.parseInt(sampling[i].trim());
  80                         }
  81                         counters = new AtomicLong[bucketUnit.length+1];
  82                         for(int i=0, max = counters.length; i<max; i++){
  83                                 counters[i] = new AtomicLong();
  84                         }
  85                         worker = new Thread(this);
  86                         worker.setDaemon(true);
  87                         worker.start();
  88                         timer = new Timer(true);
  89                         timer.schedule(new MyTimerTask(), interval*1000L, interval*1000L);
  90                 }
  91         }
  92
  93         private void register(TestSample object){
  94                 try {
  95                         tasks.add(object);
  96                 } catch(Throwable exc) {
  97                         // if cannot add an object to the queue, do nothing
  98                 }
  99         }
 100
 101         private void deRegister(TestSample object){
 102                 tasks.remove(object);
 103         }
 104
 105         public void run() {
 106                 for(;;){
 107                         Set data = new TreeSet();
 108                         tasks.drainTo(data);
 109                         for(Iterator it = data.iterator(); it.hasNext(); ){
 110                                 Object next = it.next();
 111                                 if(next instanceof TestSample){
 112                                         consume((TestSample)next);
 113                                 } else {
 114                                         System.out.println(next.getClass().getName());
 115                                 }
 116                         }
 117                         try {
 118                                 Thread.sleep(1000);
 119                         } catch (InterruptedException e) {
 120                                 e.printStackTrace();
 121                         }
 122                 }
 123
 124         }
 125         public void clearReqister(){
 126                 AtomicLong[] tmp = new AtomicLong[counters.length];
 127                 for(int i=0, max = tmp.length; i<max; i++){
 128                         tmp[i] = new AtomicLong();
 129                 }
 130                 AtomicLong[] tmp2 = counters;
 131                 synchronized(tmp2){
 132                         counters = tmp;
 133                 }
 134                 StringBuffer sb = new StringBuffer("CPM: ");
 135                 for(int i=0, max = tmp2.length; i < max; i++){
 136                         if(i==0 && bucketUnit[0]==0){
 137                                 sb.append("[Exc]=");
 138                         } else {
 139                                 sb.append("[");
 140                                 if(i==bucketUnit.length){
 141                                         sb.append("Other]=");
 142                                 } else {
 143                                         sb.append(bucketUnit[i]).append(" ms]=");
 144                                 }
 145                         }
 146                         sb.append(tmp2[i].get()).append("\t");
 147                 }
 148                 LOGGER.info(sb.toString());
 149         }
 150
 151         class MyTimerTask extends TimerTask{
 152
 153                 public void run() {
 154
 155                         clearReqister();
 156                 }
 157
 158         }
 159
 160         private void consume(TestSample probe) {
 161                 AtomicLong[] tmp = counters;
 162                 synchronized(tmp){
 163                         counters[getBucket(probe.getDuration())].incrementAndGet();
 164                 }
 165         }
 166
 167         /*
 168          * This method is used to find the offset of the bucket in
 169          * counters. 'counters' array is 1 size longer than bucketUnit,
 170          * hence by default it returns 'bucketUnit.length'
 171          */
 172         private int getBucket(long difftime){
 173                 for(int i=0; i<bucketUnit.length; i++){
 174                         if(difftime < bucketUnit[i]){
 175                                 return i;
 176                         }
 177                 }
 178                 return bucketUnit.length;
 179         }
 180
 181         private static boolean isEnabled() {
 182                 return enabled;
 183         }
 184         /**
 185          * @author Rich Tabedzki
 186          *  A helper class to pass measured parameter to the counter.
 187          */
 188         static class TestSample implements Comparable{
 189                 private long starttime;
 190                 private long endtime;
 191
 192                 public TestSample(long starttime) {
 193                         this.endtime = System.currentTimeMillis();
 194                         this.starttime = starttime;
 195                 }
 196
 197                 public long getDuration(){
 198                         return endtime - starttime;
 199                 }
 200
 201                 public int compareTo(Object o) {
 202                         if(o instanceof TestSample){
 203                                 TestSample x = (TestSample)o;
 204                                 if(starttime < x.starttime)
 205                                         return 1;
 206                                 if(endtime < x.endtime)
 207                                         return 1;
 208                                 if(starttime > x.starttime)
 209                                         return -1;
 210                                 if(endtime > x.endtime)
 211                                         return -1;
 212                                 return 0;
 213                         }
 214                         return 1;
 215                 }
 216         }
 217 }