pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImpl.java - pulsar - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package org.apache.pulsar.client.impl;

 import com.fasterxml.jackson.databind.ObjectWriter;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.yahoo.sketches.quantiles.DoublesSketch;
 import io.netty.util.Timeout;
 import io.netty.util.TimerTask;
 import java.io.IOException;
 import java.text.DecimalFormat;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;
 import org.apache.pulsar.client.api.ProducerStats;
 import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 public class ProducerStatsRecorderImpl implements ProducerStatsRecorder {

     private static final long serialVersionUID = 1L;
     private transient TimerTask stat;
     private transient Timeout statTimeout;
     private transient ProducerImpl<?> producer;
     private transient PulsarClientImpl pulsarClient;
     private long oldTime;
     private long statsIntervalSeconds;
     private final LongAdder numMsgsSent;
     private final LongAdder numBytesSent;
     private final LongAdder numSendFailed;
     private final LongAdder numAcksReceived;
     private final LongAdder totalMsgsSent;
     private final LongAdder totalBytesSent;
     private final LongAdder totalSendFailed;
     private final LongAdder totalAcksReceived;
     private static final DecimalFormat DEC = new DecimalFormat("0.000");
     private static final DecimalFormat THROUGHPUT_FORMAT = new DecimalFormat("0.00");
     private final transient DoublesSketch ds;
     private final transient DoublesSketch batchSizeDs;
     private final transient DoublesSketch msgSizeDs;

     private volatile double sendMsgsRate;
     private volatile double sendBytesRate;
     private volatile double[] latencyPctValues = new double[PERCENTILES.length];
     private volatile double[] batchSizePctValues = new double[PERCENTILES.length];
     private volatile double[] msgSizePctValues = new double[PERCENTILES.length];

     private static final double[] PERCENTILES = { 0.5, 0.75, 0.95, 0.99, 0.999, 1.0 };

     public ProducerStatsRecorderImpl() {
         numMsgsSent = new LongAdder();
         numBytesSent = new LongAdder();
         numSendFailed = new LongAdder();
         numAcksReceived = new LongAdder();
         totalMsgsSent = new LongAdder();
         totalBytesSent = new LongAdder();
         totalSendFailed = new LongAdder();
         totalAcksReceived = new LongAdder();
         ds = DoublesSketch.builder().build(256);
         batchSizeDs = DoublesSketch.builder().build(256);
         msgSizeDs = DoublesSketch.builder().build(256);
     }

     public ProducerStatsRecorderImpl(PulsarClientImpl pulsarClient, ProducerConfigurationData conf,
             ProducerImpl<?> producer) {
         this.pulsarClient = pulsarClient;
         this.statsIntervalSeconds = pulsarClient.getConfiguration().getStatsIntervalSeconds();
         this.producer = producer;
         numMsgsSent = new LongAdder();
         numBytesSent = new LongAdder();
         numSendFailed = new LongAdder();
         numAcksReceived = new LongAdder();
         totalMsgsSent = new LongAdder();
         totalBytesSent = new LongAdder();
         totalSendFailed = new LongAdder();
         totalAcksReceived = new LongAdder();
         ds = DoublesSketch.builder().build(256);
         batchSizeDs = DoublesSketch.builder().build(256);
         msgSizeDs = DoublesSketch.builder().build(256);
         init(conf);
     }

     private void init(ProducerConfigurationData conf) {
         ObjectWriter w = ObjectMapperFactory.getMapperWithIncludeAlways().writer()
                 .without(SerializationFeature.FAIL_ON_EMPTY_BEANS);

         try {
             log.info("Starting Pulsar producer perf with config: {}", w.writeValueAsString(conf));
             log.info("Pulsar client config: {}", w.writeValueAsString(pulsarClient.getConfiguration()));
         } catch (IOException e) {
             log.error("Failed to dump config info", e);
         }

         stat = (timeout) -> {

             if (timeout.isCancelled()) {
                 return;
             }

             try {
                 updateStats();
             } catch (Exception e) {
                 log.error("[{}] [{}]: {}", producer.getTopic(), producer.getProducerName(), e.getMessage());
             } finally {
                 // schedule the next stat info
                 statTimeout = pulsarClient.timer().newTimeout(stat, statsIntervalSeconds, TimeUnit.SECONDS);
             }

         };

         oldTime = System.nanoTime();
         statTimeout = pulsarClient.timer().newTimeout(stat, statsIntervalSeconds, TimeUnit.SECONDS);
     }

     Timeout getStatTimeout() {
         return statTimeout;
     }

     protected void updateStats() {
         long now = System.nanoTime();
         double elapsed = (now - oldTime) / 1e9;
         oldTime = now;

         long currentNumMsgsSent = numMsgsSent.sumThenReset();
         long currentNumBytesSent = numBytesSent.sumThenReset();
         long currentNumSendFailedMsgs = numSendFailed.sumThenReset();
         long currentNumAcksReceived = numAcksReceived.sumThenReset();

         totalMsgsSent.add(currentNumMsgsSent);
         totalBytesSent.add(currentNumBytesSent);
         totalSendFailed.add(currentNumSendFailedMsgs);
         totalAcksReceived.add(currentNumAcksReceived);

         synchronized (ds) {
             latencyPctValues = ds.getQuantiles(PERCENTILES);
             ds.reset();
         }

         synchronized (batchSizeDs) {
             batchSizePctValues = batchSizeDs.getQuantiles(PERCENTILES);
             batchSizeDs.reset();
         }

         synchronized (msgSizeDs) {
             msgSizePctValues = msgSizeDs.getQuantiles(PERCENTILES);
             msgSizeDs.reset();
         }

         sendMsgsRate = currentNumMsgsSent / elapsed;
         sendBytesRate = currentNumBytesSent / elapsed;

         if ((currentNumMsgsSent | currentNumSendFailedMsgs | currentNumAcksReceived
                 | currentNumMsgsSent) != 0) {

             for (int i = 0; i < latencyPctValues.length; i++) {
                 if (Double.isNaN(latencyPctValues[i])) {
                     latencyPctValues[i] = 0;
                 }
             }

             log.info("[{}] [{}] --- Publish throughput: {} msg/s --- {} Mbit/s --- "
                             + "Latency: med: {} ms - 95pct: {} ms - 99pct: {} ms - 99.9pct: {} ms - max: {} ms --- "
                             + "BatchSize: med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - max: {} --- "
                             + "MsgSize: med: {} bytes - 95pct: {} bytes - 99pct: {} bytes - 99.9pct: {} bytes "
                             + "- max: {} bytes --- "
                             + "Ack received rate: {} ack/s --- Failed messages: {} --- Pending messages: {}",
                     producer.getTopic(),
                     producer.getProducerName(),
                     THROUGHPUT_FORMAT.format(sendMsgsRate),
                     THROUGHPUT_FORMAT.format(sendBytesRate / 1024 / 1024 * 8),
                     DEC.format(latencyPctValues[0]), DEC.format(latencyPctValues[2]),
                     DEC.format(latencyPctValues[3]), DEC.format(latencyPctValues[4]),
                     DEC.format(latencyPctValues[5]),
                     DEC.format(batchSizePctValues[0]), DEC.format(batchSizePctValues[2]),
                     DEC.format(batchSizePctValues[3]), DEC.format(batchSizePctValues[4]),
                     DEC.format(batchSizePctValues[5]),
                     DEC.format(msgSizePctValues[0]), DEC.format(msgSizePctValues[2]),
                     DEC.format(msgSizePctValues[3]), DEC.format(msgSizePctValues[4]),
                     DEC.format(msgSizePctValues[5]),
                     THROUGHPUT_FORMAT.format(currentNumAcksReceived / elapsed), currentNumSendFailedMsgs,
                     getPendingQueueSize());
         }
     }

     @Override
     public void updateNumMsgsSent(long numMsgs, long totalMsgsSize) {
         numMsgsSent.add(numMsgs);
         numBytesSent.add(totalMsgsSize);
         synchronized (batchSizeDs) {
             batchSizeDs.update(numMsgs);
         }
         synchronized (msgSizeDs) {
             msgSizeDs.update(totalMsgsSize);
         }
     }

     @Override
     public void incrementSendFailed() {
         numSendFailed.increment();
     }

     @Override
     public void incrementSendFailed(long numMsgs) {
         numSendFailed.add(numMsgs);
     }

     @Override
     public void incrementNumAcksReceived(long latencyNs) {
         numAcksReceived.increment();
         synchronized (ds) {
             ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs));
         }
     }

     void reset() {
         numMsgsSent.reset();
         numBytesSent.reset();
         numSendFailed.reset();
         numAcksReceived.reset();
         totalMsgsSent.reset();
         totalBytesSent.reset();
         totalSendFailed.reset();
         totalAcksReceived.reset();
     }

     void updateCumulativeStats(ProducerStats stats) {
         if (stats == null) {
             return;
         }
         numMsgsSent.add(stats.getNumMsgsSent());
         numBytesSent.add(stats.getNumBytesSent());
         numSendFailed.add(stats.getNumSendFailed());
         numAcksReceived.add(stats.getNumAcksReceived());
         totalMsgsSent.add(stats.getTotalMsgsSent());
         totalBytesSent.add(stats.getTotalBytesSent());
         totalSendFailed.add(stats.getTotalSendFailed());
         totalAcksReceived.add(stats.getTotalAcksReceived());
     }

     @Override
     public long getNumMsgsSent() {
         return numMsgsSent.longValue();
     }

     @Override
     public long getNumBytesSent() {
         return numBytesSent.longValue();
     }

     @Override
     public long getNumSendFailed() {
         return numSendFailed.longValue();
     }

     @Override
     public long getNumAcksReceived() {
         return numAcksReceived.longValue();
     }

     public long getTotalMsgsSent() {
         return totalMsgsSent.longValue();
     }

     public long getTotalBytesSent() {
         return totalBytesSent.longValue();
     }

     public long getTotalSendFailed() {
         return totalSendFailed.longValue();
     }

     public long getTotalAcksReceived() {
         return totalAcksReceived.longValue();
     }

     @Override
     public double getSendMsgsRate() {
         return sendMsgsRate;
     }

     @Override
     public double getSendBytesRate() {
         return sendBytesRate;
     }

     @Override
     public double getSendLatencyMillis50pct() {
         return latencyPctValues[0];
     }

     @Override
     public double getSendLatencyMillis75pct() {
         return latencyPctValues[1];
     }

     @Override
     public double getSendLatencyMillis95pct() {
         return latencyPctValues[2];
     }

     @Override
     public double getSendLatencyMillis99pct() {
         return latencyPctValues[3];
     }

     @Override
     public double getSendLatencyMillis999pct() {
         return latencyPctValues[4];
     }

     @Override
     public double getSendLatencyMillisMax() {
         return latencyPctValues[5];
     }

     @Override
     public int getPendingQueueSize() {
         return producer.getPendingQueueSize();
     }

     public void cancelStatsTimeout() {
         this.updateStats();
         if (statTimeout != null) {
             statTimeout.cancel();
             statTimeout = null;
         }
     }

     private static final Logger log = LoggerFactory.getLogger(ProducerStatsRecorderImpl.class);
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package org.apache.pulsar.client.impl;

	import com.fasterxml.jackson.databind.ObjectWriter;
	import com.fasterxml.jackson.databind.SerializationFeature;
	import com.yahoo.sketches.quantiles.DoublesSketch;
	import io.netty.util.Timeout;
	import io.netty.util.TimerTask;
	import java.io.IOException;
	import java.text.DecimalFormat;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.LongAdder;
	import org.apache.pulsar.client.api.ProducerStats;
	import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
	import org.apache.pulsar.common.util.ObjectMapperFactory;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	public class ProducerStatsRecorderImpl implements ProducerStatsRecorder {

	private static final long serialVersionUID = 1L;
	private transient TimerTask stat;
	private transient Timeout statTimeout;
	private transient ProducerImpl<?> producer;
	private transient PulsarClientImpl pulsarClient;
	private long oldTime;
	private long statsIntervalSeconds;
	private final LongAdder numMsgsSent;
	private final LongAdder numBytesSent;
	private final LongAdder numSendFailed;
	private final LongAdder numAcksReceived;
	private final LongAdder totalMsgsSent;
	private final LongAdder totalBytesSent;
	private final LongAdder totalSendFailed;
	private final LongAdder totalAcksReceived;
	private static final DecimalFormat DEC = new DecimalFormat("0.000");
	private static final DecimalFormat THROUGHPUT_FORMAT = new DecimalFormat("0.00");
	private final transient DoublesSketch ds;
	private final transient DoublesSketch batchSizeDs;
	private final transient DoublesSketch msgSizeDs;

	private volatile double sendMsgsRate;
	private volatile double sendBytesRate;
	private volatile double[] latencyPctValues = new double[PERCENTILES.length];
	private volatile double[] batchSizePctValues = new double[PERCENTILES.length];
	private volatile double[] msgSizePctValues = new double[PERCENTILES.length];

	private static final double[] PERCENTILES = { 0.5, 0.75, 0.95, 0.99, 0.999, 1.0 };

	public ProducerStatsRecorderImpl() {
	numMsgsSent = new LongAdder();
	numBytesSent = new LongAdder();
	numSendFailed = new LongAdder();
	numAcksReceived = new LongAdder();
	totalMsgsSent = new LongAdder();
	totalBytesSent = new LongAdder();
	totalSendFailed = new LongAdder();
	totalAcksReceived = new LongAdder();
	ds = DoublesSketch.builder().build(256);
	batchSizeDs = DoublesSketch.builder().build(256);
	msgSizeDs = DoublesSketch.builder().build(256);
	}

	public ProducerStatsRecorderImpl(PulsarClientImpl pulsarClient, ProducerConfigurationData conf,
	ProducerImpl<?> producer) {
	this.pulsarClient = pulsarClient;
	this.statsIntervalSeconds = pulsarClient.getConfiguration().getStatsIntervalSeconds();
	this.producer = producer;
	numMsgsSent = new LongAdder();
	numBytesSent = new LongAdder();
	numSendFailed = new LongAdder();
	numAcksReceived = new LongAdder();
	totalMsgsSent = new LongAdder();
	totalBytesSent = new LongAdder();
	totalSendFailed = new LongAdder();
	totalAcksReceived = new LongAdder();
	ds = DoublesSketch.builder().build(256);
	batchSizeDs = DoublesSketch.builder().build(256);
	msgSizeDs = DoublesSketch.builder().build(256);
	init(conf);
	}

	private void init(ProducerConfigurationData conf) {
	ObjectWriter w = ObjectMapperFactory.getMapperWithIncludeAlways().writer()
	.without(SerializationFeature.FAIL_ON_EMPTY_BEANS);

	try {
	log.info("Starting Pulsar producer perf with config: {}", w.writeValueAsString(conf));
	log.info("Pulsar client config: {}", w.writeValueAsString(pulsarClient.getConfiguration()));
	} catch (IOException e) {
	log.error("Failed to dump config info", e);
	}

	stat = (timeout) -> {

	if (timeout.isCancelled()) {
	return;
	}

	try {
	updateStats();
	} catch (Exception e) {
	log.error("[{}] [{}]: {}", producer.getTopic(), producer.getProducerName(), e.getMessage());
	} finally {
	// schedule the next stat info
	statTimeout = pulsarClient.timer().newTimeout(stat, statsIntervalSeconds, TimeUnit.SECONDS);
	}

	};

	oldTime = System.nanoTime();
	statTimeout = pulsarClient.timer().newTimeout(stat, statsIntervalSeconds, TimeUnit.SECONDS);
	}

	Timeout getStatTimeout() {
	return statTimeout;
	}

	protected void updateStats() {
	long now = System.nanoTime();
	double elapsed = (now - oldTime) / 1e9;
	oldTime = now;

	long currentNumMsgsSent = numMsgsSent.sumThenReset();
	long currentNumBytesSent = numBytesSent.sumThenReset();
	long currentNumSendFailedMsgs = numSendFailed.sumThenReset();
	long currentNumAcksReceived = numAcksReceived.sumThenReset();

	totalMsgsSent.add(currentNumMsgsSent);
	totalBytesSent.add(currentNumBytesSent);
	totalSendFailed.add(currentNumSendFailedMsgs);
	totalAcksReceived.add(currentNumAcksReceived);

	synchronized (ds) {
	latencyPctValues = ds.getQuantiles(PERCENTILES);
	ds.reset();
	}

	synchronized (batchSizeDs) {
	batchSizePctValues = batchSizeDs.getQuantiles(PERCENTILES);
	batchSizeDs.reset();
	}

	synchronized (msgSizeDs) {
	msgSizePctValues = msgSizeDs.getQuantiles(PERCENTILES);
	msgSizeDs.reset();
	}

	sendMsgsRate = currentNumMsgsSent / elapsed;
	sendBytesRate = currentNumBytesSent / elapsed;

	if ((currentNumMsgsSent \| currentNumSendFailedMsgs \| currentNumAcksReceived
	\| currentNumMsgsSent) != 0) {

	for (int i = 0; i < latencyPctValues.length; i++) {
	if (Double.isNaN(latencyPctValues[i])) {
	latencyPctValues[i] = 0;
	}
	}

	log.info("[{}] [{}] --- Publish throughput: {} msg/s --- {} Mbit/s --- "
	+ "Latency: med: {} ms - 95pct: {} ms - 99pct: {} ms - 99.9pct: {} ms - max: {} ms --- "
	+ "BatchSize: med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - max: {} --- "
	+ "MsgSize: med: {} bytes - 95pct: {} bytes - 99pct: {} bytes - 99.9pct: {} bytes "
	+ "- max: {} bytes --- "
	+ "Ack received rate: {} ack/s --- Failed messages: {} --- Pending messages: {}",
	producer.getTopic(),
	producer.getProducerName(),
	THROUGHPUT_FORMAT.format(sendMsgsRate),
	THROUGHPUT_FORMAT.format(sendBytesRate / 1024 / 1024 * 8),
	DEC.format(latencyPctValues[0]), DEC.format(latencyPctValues[2]),
	DEC.format(latencyPctValues[3]), DEC.format(latencyPctValues[4]),
	DEC.format(latencyPctValues[5]),
	DEC.format(batchSizePctValues[0]), DEC.format(batchSizePctValues[2]),
	DEC.format(batchSizePctValues[3]), DEC.format(batchSizePctValues[4]),
	DEC.format(batchSizePctValues[5]),
	DEC.format(msgSizePctValues[0]), DEC.format(msgSizePctValues[2]),
	DEC.format(msgSizePctValues[3]), DEC.format(msgSizePctValues[4]),
	DEC.format(msgSizePctValues[5]),
	THROUGHPUT_FORMAT.format(currentNumAcksReceived / elapsed), currentNumSendFailedMsgs,
	getPendingQueueSize());
	}
	}

	@Override
	public void updateNumMsgsSent(long numMsgs, long totalMsgsSize) {
	numMsgsSent.add(numMsgs);
	numBytesSent.add(totalMsgsSize);
	synchronized (batchSizeDs) {
	batchSizeDs.update(numMsgs);
	}
	synchronized (msgSizeDs) {
	msgSizeDs.update(totalMsgsSize);
	}
	}

	@Override
	public void incrementSendFailed() {
	numSendFailed.increment();
	}

	@Override
	public void incrementSendFailed(long numMsgs) {
	numSendFailed.add(numMsgs);
	}

	@Override
	public void incrementNumAcksReceived(long latencyNs) {
	numAcksReceived.increment();
	synchronized (ds) {
	ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs));
	}
	}

	void reset() {
	numMsgsSent.reset();
	numBytesSent.reset();
	numSendFailed.reset();
	numAcksReceived.reset();
	totalMsgsSent.reset();
	totalBytesSent.reset();
	totalSendFailed.reset();
	totalAcksReceived.reset();
	}

	void updateCumulativeStats(ProducerStats stats) {
	if (stats == null) {
	return;
	}
	numMsgsSent.add(stats.getNumMsgsSent());
	numBytesSent.add(stats.getNumBytesSent());
	numSendFailed.add(stats.getNumSendFailed());
	numAcksReceived.add(stats.getNumAcksReceived());
	totalMsgsSent.add(stats.getTotalMsgsSent());
	totalBytesSent.add(stats.getTotalBytesSent());
	totalSendFailed.add(stats.getTotalSendFailed());
	totalAcksReceived.add(stats.getTotalAcksReceived());
	}

	@Override
	public long getNumMsgsSent() {
	return numMsgsSent.longValue();
	}

	@Override
	public long getNumBytesSent() {
	return numBytesSent.longValue();
	}

	@Override
	public long getNumSendFailed() {
	return numSendFailed.longValue();
	}

	@Override
	public long getNumAcksReceived() {
	return numAcksReceived.longValue();
	}

	public long getTotalMsgsSent() {
	return totalMsgsSent.longValue();
	}

	public long getTotalBytesSent() {
	return totalBytesSent.longValue();
	}

	public long getTotalSendFailed() {
	return totalSendFailed.longValue();
	}

	public long getTotalAcksReceived() {
	return totalAcksReceived.longValue();
	}

	@Override
	public double getSendMsgsRate() {
	return sendMsgsRate;
	}

	@Override
	public double getSendBytesRate() {
	return sendBytesRate;
	}

	@Override
	public double getSendLatencyMillis50pct() {
	return latencyPctValues[0];
	}

	@Override
	public double getSendLatencyMillis75pct() {
	return latencyPctValues[1];
	}

	@Override
	public double getSendLatencyMillis95pct() {
	return latencyPctValues[2];
	}

	@Override
	public double getSendLatencyMillis99pct() {
	return latencyPctValues[3];
	}

	@Override
	public double getSendLatencyMillis999pct() {
	return latencyPctValues[4];
	}

	@Override
	public double getSendLatencyMillisMax() {
	return latencyPctValues[5];
	}

	@Override
	public int getPendingQueueSize() {
	return producer.getPendingQueueSize();
	}

	public void cancelStatsTimeout() {
	this.updateStats();
	if (statTimeout != null) {
	statTimeout.cancel();
	statTimeout = null;
	}
	}

	private static final Logger log = LoggerFactory.getLogger(ProducerStatsRecorderImpl.class);
	}