src/main/java/com/att/nsa/cambria/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   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  *        http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package com.att.nsa.cambria.metrics.publisher.impl;
  23
  24 import java.io.ByteArrayOutputStream;
  25 import java.io.IOException;
  26 import java.io.OutputStream;
  27 import java.net.MalformedURLException;
  28 import java.util.Collection;
  29 import java.util.LinkedList;
  30 import java.util.List;
  31 import java.util.concurrent.LinkedBlockingQueue;
  32 import java.util.concurrent.ScheduledThreadPoolExecutor;
  33 import java.util.concurrent.TimeUnit;
  34 import java.util.zip.GZIPOutputStream;
  35
  36 import javax.ws.rs.client.Client;
  37 import javax.ws.rs.client.ClientBuilder;
  38 import javax.ws.rs.client.Entity;
  39 import javax.ws.rs.client.WebTarget;
  40 import javax.ws.rs.core.Response;
  41
  42 import com.att.ajsc.filemonitor.AJSCPropertiesMap;
  43 import com.att.nsa.cambria.constants.CambriaConstants;
  44 import com.att.nsa.cambria.metrics.publisher.CambriaPublisherUtility;
  45
  46 /**
  47  *
  48  * class DMaaPCambriaSimplerBatchPublisher used to send the publish the messages
  49  * in batch
  50  *
  51  * @author author
  52  *
  53  */
  54 public class DMaaPCambriaSimplerBatchPublisher extends CambriaBaseClient
  55                 implements com.att.nsa.cambria.metrics.publisher.CambriaBatchingPublisher {
  56         /**
  57          *
  58          * static inner class initializes with urls, topic,batchSize
  59          *
  60          * @author author
  61          *
  62          */
  63         public static class Builder {
  64                 public Builder() {
  65                 }
  66
  67                 /**
  68                  * constructor initialize with url
  69                  *
  70                  * @param baseUrls
  71                  * @return
  72                  *
  73                  */
  74                 public Builder againstUrls(Collection<String> baseUrls) {
  75                         fUrls = baseUrls;
  76                         return this;
  77                 }
  78
  79                 /**
  80                  * constructor initializes with topics
  81                  *
  82                  * @param topic
  83                  * @return
  84                  *
  85                  */
  86                 public Builder onTopic(String topic) {
  87                         fTopic = topic;
  88                         return this;
  89                 }
  90
  91                 /**
  92                  * constructor initilazes with batch size and batch time
  93                  *
  94                  * @param maxBatchSize
  95                  * @param maxBatchAgeMs
  96                  * @return
  97                  *
  98                  */
  99                 public Builder batchTo(int maxBatchSize, long maxBatchAgeMs) {
 100                         fMaxBatchSize = maxBatchSize;
 101                         fMaxBatchAgeMs = maxBatchAgeMs;
 102                         return this;
 103                 }
 104
 105                 /**
 106                  * constructor initializes with compress
 107                  *
 108                  * @param compress
 109                  * @return
 110                  */
 111                 public Builder compress(boolean compress) {
 112                         fCompress = compress;
 113                         return this;
 114                 }
 115
 116                 /**
 117                  * method returns DMaaPCambriaSimplerBatchPublisher object
 118                  *
 119                  * @return
 120                  */
 121                 public DMaaPCambriaSimplerBatchPublisher build() {
 122                         try {
 123                                 return new DMaaPCambriaSimplerBatchPublisher(fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress);
 124                         } catch (MalformedURLException e) {
 125                                 throw new RuntimeException(e);
 126                         }
 127                 }
 128
 129                 private Collection<String> fUrls;
 130                 private String fTopic;
 131                 private int fMaxBatchSize = 100;
 132                 private long fMaxBatchAgeMs = 1000;
 133                 private boolean fCompress = false;
 134         };
 135
 136         /**
 137          *
 138          * @param partition
 139          * @param msg
 140          */
 141         @Override
 142         public int send(String partition, String msg) {
 143                 return send(new message(partition, msg));
 144         }
 145
 146         /**
 147          * @param msg
 148          */
 149         @Override
 150         public int send(message msg) {
 151                 final LinkedList<message> list = new LinkedList<message>();
 152                 list.add(msg);
 153                 return send(list);
 154         }
 155
 156         /**
 157          * @param msgs
 158          */
 159         @Override
 160         public synchronized int send(Collection<message> msgs) {
 161                 if (fClosed) {
 162                         throw new IllegalStateException("The publisher was closed.");
 163                 }
 164
 165                 for (message userMsg : msgs) {
 166                         fPending.add(new TimestampedMessage(userMsg));
 167                 }
 168                 return getPendingMessageCount();
 169         }
 170
 171         /**
 172          * getPending message count
 173          */
 174         @Override
 175         public synchronized int getPendingMessageCount() {
 176                 return fPending.size();
 177         }
 178
 179         /**
 180          *
 181          * @exception InterruptedException
 182          * @exception IOException
 183          */
 184         @Override
 185         public void close() {
 186                 try {
 187                         final List<message> remains = close(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
 188                         if (remains.size() > 0) {
 189                                 getLog().warn("Closing publisher with " + remains.size() + " messages unsent. "
 190                                                 + "Consider using CambriaBatchingPublisher.close( long timeout, TimeUnit timeoutUnits ) to recapture unsent messages on close.");
 191                         }
 192                 } catch (InterruptedException e) {
 193                         getLog().info(" Interruption Exception is caught here : " + e.getMessage());
 194                         Thread.currentThread().interrupt();
 195                 } catch (IOException e) {
 196                         getLog().warn("Possible message loss. " + e.getMessage(), e);
 197                 }
 198         }
 199
 200         /**
 201          * @param time
 202          * @param unit
 203          */
 204         @Override
 205         public List<message> close(long time, TimeUnit unit) throws IOException, InterruptedException {
 206                 synchronized (this) {
 207                         fClosed = true;
 208
 209                         // stop the background sender
 210                         fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
 211                         fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
 212                         fExec.shutdown();
 213                 }
 214
 215                 final long now = Clock.now();
 216                 final long waitInMs = TimeUnit.MILLISECONDS.convert(time, unit);
 217                 final long timeoutAtMs = now + waitInMs;
 218
 219                 while (Clock.now() < timeoutAtMs && getPendingMessageCount() > 0) {
 220                         send(true);
 221                         Thread.sleep(250);
 222                 }
 223                 // synchronizing the current object
 224                 synchronized (this) {
 225                         final LinkedList<message> result = new LinkedList<message>();
 226                         fPending.drainTo(result);
 227                         return result;
 228                 }
 229         }
 230
 231         /**
 232          * Possibly send a batch to the cambria server. This is called by the
 233          * background thread and the close() method
 234          *
 235          * @param force
 236          */
 237         private synchronized void send(boolean force) {
 238                 if (force || shouldSendNow()) {
 239                         if (!sendBatch()) {
 240                                 getLog().warn("Send failed, " + fPending.size() + " message to send.");
 241
 242                                 // note the time for back-off
 243                                 fDontSendUntilMs = sfWaitAfterError + Clock.now();
 244                         }
 245                 }
 246         }
 247
 248         /**
 249          *
 250          * @return
 251          */
 252         private synchronized boolean shouldSendNow() {
 253                 boolean shouldSend = false;
 254                 if (fPending.size() > 0) {
 255                         final long nowMs = Clock.now();
 256
 257                         shouldSend = (fPending.size() >= fMaxBatchSize);
 258                         if (!shouldSend) {
 259                                 final long sendAtMs = fPending.peek().timestamp + fMaxBatchAgeMs;
 260                                 shouldSend = sendAtMs <= nowMs;
 261                         }
 262
 263                         // however, wait after an error
 264                         shouldSend = shouldSend && nowMs >= fDontSendUntilMs;
 265                 }
 266                 return shouldSend;
 267         }
 268
 269         /**
 270          *
 271          * @return
 272          */
 273         private synchronized boolean sendBatch() {
 274                 // it's possible for this call to be made with an empty list. in this
 275                 // case, just return.
 276                 if (fPending.size() < 1) {
 277                         return true;
 278                 }
 279
 280                 final long nowMs = Clock.now();
 281                 final String url = CambriaPublisherUtility.makeUrl(fTopic);
 282
 283                 getLog().info("sending " + fPending.size() + " msgs to " + url + ". Oldest: "
 284                                 + (nowMs - fPending.peek().timestamp) + " ms");
 285
 286                 try {
 287
 288                         final ByteArrayOutputStream baseStream = new ByteArrayOutputStream();
 289                         OutputStream os = baseStream;
 290                         if (fCompress) {
 291                                 os = new GZIPOutputStream(baseStream);
 292                         }
 293                         for (TimestampedMessage m : fPending) {
 294                                 os.write(("" + m.fPartition.length()).getBytes());
 295                                 os.write('.');
 296                                 os.write(("" + m.fMsg.length()).getBytes());
 297                                 os.write('.');
 298                                 os.write(m.fPartition.getBytes());
 299                                 os.write(m.fMsg.getBytes());
 300                                 os.write('\n');
 301                         }
 302                         os.close();
 303
 304                         final long startMs = Clock.now();
 305
 306                         // code from REST Client Starts
 307
 308                         // final String serverCalculatedSignature = sha1HmacSigner.sign
 309                         // ("2015-09-21T11:38:19-0700", "iHAxArrj6Ve9JgmHvR077QiV");
 310
 311                         Client client = ClientBuilder.newClient();
 312                         String metricTopicname = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"metrics.send.cambria.topic");
 313                          if (null==metricTopicname) {
 314
 315                          metricTopicname="msgrtr.apinode.metrics.dmaap";
 316                          }
 317                         WebTarget target = client
 318                                         .target("http://localhost:" + CambriaConstants.kStdCambriaServicePort);
 319                         target = target.path("/events/" + fTopic);
 320                         getLog().info("url : " + target.getUri().toString());
 321                         // API Key
 322
 323                         Entity<byte[]> data = Entity.entity(baseStream.toByteArray(), "application/cambria");
 324
 325                         Response response = target.request().post(data);
 326                         // header("X-CambriaAuth",
 327                         // "2OH46YIWa329QpEF:"+serverCalculatedSignature).
 328                         // header("X-CambriaDate", "2015-09-21T11:38:19-0700").
 329                         // post(Entity.json(baseStream.toByteArray()));
 330
 331                         getLog().info("Response received :: " + response.getStatus());
 332                         getLog().info("Response received :: " + response.toString());
 333
 334                         // code from REST Client Ends
 335
 336                         /*
 337                          * final JSONObject result = post ( url, contentType,
 338                          * baseStream.toByteArray(), true ); final String logLine =
 339                          * "cambria reply ok (" + (Clock.now()-startMs) + " ms):" +
 340                          * result.toString (); getLog().info ( logLine );
 341                          */
 342                         fPending.clear();
 343                         return true;
 344                 } catch (IllegalArgumentException x) {
 345                         getLog().warn(x.getMessage(), x);
 346                 }
 347                 /*
 348                  * catch ( HttpObjectNotFoundException x ) { getLog().warn (
 349                  * x.getMessage(), x ); } catch ( HttpException x ) { getLog().warn (
 350                  * x.getMessage(), x ); }
 351                  */
 352                 catch (IOException x) {
 353                         getLog().warn(x.getMessage(), x);
 354                 }
 355                 return false;
 356         }
 357
 358         private final String fTopic;
 359         private final int fMaxBatchSize;
 360         private final long fMaxBatchAgeMs;
 361         private final boolean fCompress;
 362         private boolean fClosed;
 363
 364         private final LinkedBlockingQueue<TimestampedMessage> fPending;
 365         private long fDontSendUntilMs;
 366         private final ScheduledThreadPoolExecutor fExec;
 367
 368         private static final long sfWaitAfterError = 1000;
 369
 370         /**
 371          *
 372          * @param hosts
 373          * @param topic
 374          * @param maxBatchSize
 375          * @param maxBatchAgeMs
 376          * @param compress
 377          * @throws MalformedURLException
 378          */
 379         private DMaaPCambriaSimplerBatchPublisher(Collection<String> hosts, String topic, int maxBatchSize,
 380                         long maxBatchAgeMs, boolean compress) throws MalformedURLException {
 381
 382                 super(hosts);
 383
 384                 if (topic == null || topic.length() < 1) {
 385                         throw new IllegalArgumentException("A topic must be provided.");
 386                 }
 387
 388                 fClosed = false;
 389                 fTopic = topic;
 390                 fMaxBatchSize = maxBatchSize;
 391                 fMaxBatchAgeMs = maxBatchAgeMs;
 392                 fCompress = compress;
 393
 394                 fPending = new LinkedBlockingQueue<TimestampedMessage>();
 395                 fDontSendUntilMs = 0;
 396
 397                 fExec = new ScheduledThreadPoolExecutor(1);
 398                 fExec.scheduleAtFixedRate(new Runnable() {
 399                         @Override
 400                         public void run() {
 401                                 send(false);
 402                         }
 403                 }, 100, 50, TimeUnit.MILLISECONDS);
 404         }
 405
 406         /**
 407          *
 408          *
 409          * @author author
 410          *
 411          */
 412         private static class TimestampedMessage extends message {
 413                 /**
 414                  * to store timestamp value
 415                  */
 416                 public final long timestamp;
 417
 418                 /**
 419                  * constructor initialize with message
 420                  *
 421                  * @param m
 422                  *
 423                  */
 424                 public TimestampedMessage(message m) {
 425                         super(m);
 426                         timestamp = Clock.now();
 427                 }
 428         }
 429
 430 }