Java 类com.amazonaws.services.cloudwatch.model.MetricDatum 实例源码

/**
 * Flush all the current aggregated MetricDatum and return as a list.
 * This is safe to call concurrently with {@link #add}.
 * All data added prior to a flush call will be included in the returned aggregate.
 * Any data added after the flush call returns will be included in a subsequent flush.
 * Data added while a flush call is processing may be included in the current flush
 * or a subsequent flush, but will not be included twice.
 *
 * The timestamp on the aggregated data will be the time it was flushed,
 * not the time of any of the original metric events.
 *
 * @return list of all data aggregated since the last flush
 */
public List<MetricDatum> flush() {
    if (statisticsMap.size() == 0) {
        return Collections.emptyList();
    }

    // Capture all the current metrics, as represented by the set of keys
    // at this time in the statisticsMap.
    // Note that this iterates over the key set of the underlying map, and
    // removes keys from the map at the same time. It is possible keys may
    // be added during this iteration, or data for keys modified between
    // a key being chosen for iteration and being removed from the map.
    // This is ok.  Any new keys will be picked up on subsequent flushes.
    //TODO: use two maps and swap between, to ensure 'perfect' segmentation?
    List<MetricDatum> metricData = new ArrayList<>();
    for (DatumKey key : statisticsMap.keySet()) {
        StatisticSet value = statisticsMap.remove(key);
        //TODO: better to have no timestamp at all?
        MetricDatum metricDatum = key.getDatum().withTimestamp(Date.from(Instant.now()))
                                                .withStatisticValues(value);
        metricData.add(metricDatum);
    }

    return metricData;
}

private void sendAggregatedData() {

        // Grab all the current aggregated data, resetting
        // the aggregator to empty in the process
        List<MetricDatum> metricData = aggregator.flush();
        if(metricData.isEmpty()) {
            return;
        }

        // Send the data in batches to adhere to CloudWatch limitation on the
        // number of MetricDatum objects per request, see:
        // http://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/cloudwatch_limits.html
        int begin = 0;
        while(begin < metricData.size()) {
            int end = begin + BATCH_SIZE;

            sendData(metricData.subList(begin, Math.min(end, metricData.size())));

            begin = end;
        }

    }

private MetricDatum makeDatum(
        final String id,
        final String name,
        final double sum,
        final double min,
        final double max,
        final int count,
        final StandardUnit unit)
{
    MetricDatum md = new MetricDatum().withMetricName(name).withUnit(unit);

    final StatisticSet statSet = new StatisticSet()
            .withSampleCount(Double.valueOf(count))
            .withSum(sum)
            .withMinimum(min)
            .withMaximum(max);
    md.setStatisticValues(statSet);

    List<Dimension> dimensions = new ArrayList<>(1);
    Dimension trace = new Dimension().withName(ContextData.ID.name).withValue(id);

    dimensions.add(trace);
    md.setDimensions(dimensions);

    return md;
}

/**
 * The {@link Snapshot} values of {@link Timer} are reported as {@link StatisticSet} after conversion. The
 * conversion is done using the duration factor, which is deduced from the set duration unit.
 * <p>
 * Please note, the reported values submitted only if they show some data (greater than zero) in order to:
 * <p>
 * 1. save some money
 * 2. prevent com.amazonaws.services.cloudwatch.model.InvalidParameterValueException if empty {@link Snapshot}
 * is submitted
 * <p>
 * If {@link Builder#withZeroValuesSubmission()} is {@code true}, then all values will be submitted
 *
 * @see Timer#getSnapshot
 * @see #getDurationUnit
 * @see #convertDuration(double)
 */
private void processTimer(final String metricName, final Timer timer, final List<MetricDatum> metricData) {
    final Snapshot snapshot = timer.getSnapshot();

    if (builder.withZeroValuesSubmission || snapshot.size() > 0) {
        for (final Percentile percentile : builder.percentiles) {
            final double convertedDuration = convertDuration(snapshot.getValue(percentile.getQuantile()));
            stageMetricDatum(true, metricName, convertedDuration, durationUnit, percentile.getDesc(), metricData);
        }
    }

    // prevent empty snapshot from causing InvalidParameterValueException
    if (snapshot.size() > 0) {
        final String formattedDuration = String.format(" [in-%s]", getDurationUnit());
        stageMetricDatum(builder.withArithmeticMean, metricName, convertDuration(snapshot.getMean()), durationUnit, DIMENSION_SNAPSHOT_MEAN + formattedDuration, metricData);
        stageMetricDatum(builder.withStdDev, metricName, convertDuration(snapshot.getStdDev()), durationUnit, DIMENSION_SNAPSHOT_STD_DEV + formattedDuration, metricData);
        stageMetricDatumWithConvertedSnapshot(builder.withStatisticSet, metricName, snapshot, durationUnit, metricData);
    }
}

/**
 * The {@link Snapshot} values of {@link Histogram} are reported as {@link StatisticSet} raw. In other words, the
 * conversion using the duration factor does NOT apply.
 * <p>
 * Please note, the reported values submitted only if they show some data (greater than zero) in order to:
 * <p>
 * 1. save some money
 * 2. prevent com.amazonaws.services.cloudwatch.model.InvalidParameterValueException if empty {@link Snapshot}
 * is submitted
 * <p>
 * If {@link Builder#withZeroValuesSubmission()} is {@code true}, then all values will be submitted
 *
 * @see Histogram#getSnapshot
 */
private void processHistogram(final String metricName, final Histogram histogram, final List<MetricDatum> metricData) {
    final Snapshot snapshot = histogram.getSnapshot();

    if (builder.withZeroValuesSubmission || snapshot.size() > 0) {
        for (final Percentile percentile : builder.percentiles) {
            final double value = snapshot.getValue(percentile.getQuantile());
            stageMetricDatum(true, metricName, value, StandardUnit.None, percentile.getDesc(), metricData);
        }
    }

    // prevent empty snapshot from causing InvalidParameterValueException
    if (snapshot.size() > 0) {
        stageMetricDatum(builder.withArithmeticMean, metricName, snapshot.getMean(), StandardUnit.None, DIMENSION_SNAPSHOT_MEAN, metricData);
        stageMetricDatum(builder.withStdDev, metricName, snapshot.getStdDev(), StandardUnit.None, DIMENSION_SNAPSHOT_STD_DEV, metricData);
        stageMetricDatumWithRawSnapshot(builder.withStatisticSet, metricName, snapshot, StandardUnit.None, metricData);
    }
}

/**
 * Please note, the reported values submitted only if they show some data (greater than zero) in order to:
 * <p>
 * 1. save some money
 * 2. prevent com.amazonaws.services.cloudwatch.model.InvalidParameterValueException if empty {@link Snapshot}
 * is submitted
 * <p>
 * If {@link Builder#withZeroValuesSubmission()} is {@code true}, then all values will be submitted
 */
private void stageMetricDatum(final boolean metricConfigured,
                              final String metricName,
                              final double metricValue,
                              final StandardUnit standardUnit,
                              final String dimensionValue,
                              final List<MetricDatum> metricData) {
    // Only submit metrics that show some data, so let's save some money
    if (metricConfigured && (builder.withZeroValuesSubmission || metricValue > 0)) {
        final Set<Dimension> dimensions = new LinkedHashSet<>(builder.globalDimensions);
        dimensions.add(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(dimensionValue));

        metricData.add(new MetricDatum()
                .withTimestamp(new Date(builder.clock.getTime()))
                .withValue(cleanMetricValue(metricValue))
                .withMetricName(metricName)
                .withDimensions(dimensions)
                .withUnit(standardUnit));
    }
}

private void stageMetricDatumWithConvertedSnapshot(final boolean metricConfigured,
                                                   final String metricName,
                                                   final Snapshot snapshot,
                                                   final StandardUnit standardUnit,
                                                   final List<MetricDatum> metricData) {
    if (metricConfigured) {
        double scaledSum = convertDuration(LongStream.of(snapshot.getValues()).sum());
        final StatisticSet statisticSet = new StatisticSet()
                .withSum(scaledSum)
                .withSampleCount((double) snapshot.size())
                .withMinimum(convertDuration(snapshot.getMin()))
                .withMaximum(convertDuration(snapshot.getMax()));

        final Set<Dimension> dimensions = new LinkedHashSet<>(builder.globalDimensions);
        dimensions.add(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(DIMENSION_SNAPSHOT_SUMMARY));

        metricData.add(new MetricDatum()
                .withTimestamp(new Date(builder.clock.getTime()))
                .withMetricName(metricName)
                .withDimensions(dimensions)
                .withStatisticValues(statisticSet)
                .withUnit(standardUnit));
    }
}

private void stageMetricDatumWithRawSnapshot(final boolean metricConfigured,
                                             final String metricName,
                                             final Snapshot snapshot,
                                             final StandardUnit standardUnit,
                                             final List<MetricDatum> metricData) {
    if (metricConfigured) {
        double total = LongStream.of(snapshot.getValues()).sum();
        final StatisticSet statisticSet = new StatisticSet()
                .withSum(total)
                .withSampleCount((double) snapshot.size())
                .withMinimum((double) snapshot.getMin())
                .withMaximum((double) snapshot.getMax());

        final Set<Dimension> dimensions = new LinkedHashSet<>(builder.globalDimensions);
        dimensions.add(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(DIMENSION_SNAPSHOT_SUMMARY));

        metricData.add(new MetricDatum()
                .withTimestamp(new Date(builder.clock.getTime()))
                .withMetricName(metricName)
                .withDimensions(dimensions)
                .withStatisticValues(statisticSet)
                .withUnit(standardUnit));
    }
}

@Test
public void reportMetersCountersGaugesWithZeroValuesOnlyWhenConfigured() throws Exception {
    metricRegistry.register(ARBITRARY_GAUGE_NAME, (Gauge<Long>) () -> 0L);
    metricRegistry.meter(ARBITRARY_METER_NAME).mark(0);
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc(0);
    metricRegistry.timer(ARBITRARY_TIMER_NAME).update(-1L, TimeUnit.NANOSECONDS);

    buildReportWithSleep(reporterBuilder
            .withArithmeticMean()
            .withOneMinuteMeanRate()
            .withFiveMinuteMeanRate()
            .withFifteenMinuteMeanRate()
            .withZeroValuesSubmission()
            .withMeanRate());

    verify(mockAmazonCloudWatchAsyncClient, times(1)).putMetricDataAsync(metricDataRequestCaptor.capture());

    final PutMetricDataRequest putMetricDataRequest = metricDataRequestCaptor.getValue();
    final List<MetricDatum> metricData = putMetricDataRequest.getMetricData();
    for (final MetricDatum metricDatum : metricData) {
        assertThat(metricDatum.getValue()).isEqualTo(0.0);
    }
}

@Test
public void shouldReportCounterValueDelta() throws Exception {
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    final CloudWatchReporter cloudWatchReporter = reporterBuilder.build();

    cloudWatchReporter.report();
    MetricDatum metricDatum = firstMetricDatumFromCapturedRequest();
    assertThat(metricDatum.getValue().intValue()).isEqualTo(2);
    metricDataRequestCaptor.getAllValues().clear();

    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();

    cloudWatchReporter.report();
    metricDatum = firstMetricDatumFromCapturedRequest();
    assertThat(metricDatum.getValue().intValue()).isEqualTo(6);

    verify(mockAmazonCloudWatchAsyncClient, times(2)).putMetricDataAsync(any(PutMetricDataRequest.class));
}

@Test
public void shouldReportSnapshotValuesWithoutConversionWhenReportingHistogram() throws Exception {
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(1);
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(2);
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(3);
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(30);
    reporterBuilder.withStatisticSet().build().report();

    final MetricDatum metricData = metricDatumByDimensionFromCapturedRequest(DIMENSION_SNAPSHOT_SUMMARY);

    assertThat(metricData.getStatisticValues().getSum().intValue()).isEqualTo(36);
    assertThat(metricData.getStatisticValues().getMaximum().intValue()).isEqualTo(30);
    assertThat(metricData.getStatisticValues().getMinimum().intValue()).isEqualTo(1);
    assertThat(metricData.getStatisticValues().getSampleCount().intValue()).isEqualTo(4);
    assertThat(metricData.getUnit()).isEqualTo(None.toString());
}

private MetricDatum metricDatumByDimensionFromCapturedRequest(final String dimensionValue) {
    final PutMetricDataRequest putMetricDataRequest = metricDataRequestCaptor.getValue();
    final List<MetricDatum> metricData = putMetricDataRequest.getMetricData();

    final Optional<MetricDatum> metricDatumOptional =
            metricData
                    .stream()
                    .filter(metricDatum -> metricDatum.getDimensions()
                            .contains(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(dimensionValue)))
                    .findFirst();

    if (metricDatumOptional.isPresent()) {
        return metricDatumOptional.get();
    }

    throw new IllegalStateException("Could not find MetricDatum for Dimension value: " + dimensionValue);
}

@Override
protected void runOneIteration() throws Exception {
    try {
        _cloudWatch.putMetricData(
                new PutMetricDataRequest()
                        .withNamespace(NAMESPACE)
                        .withMetricData(
                                new MetricDatum()
                                        .withTimestamp(new Date())
                                        .withMetricName(ACTIVE_AND_PENDING_SCANS)
                                        .withValue((double) (_activeScanCount + _pendingScanCount))
                                        .withUnit(StandardUnit.Count)
                                        .withDimensions(_dimensions)));
    } catch (AmazonClientException e) {
        _log.error("Failed to publish active and pending scans metric", e);
    }
}

private List<MetricDatum> getMetricData(Exchange exchange) {
    Object body = exchange.getIn().getBody();
    if (body instanceof List) {
        return CastUtils.cast((List<?>) body);
    }

    if (body instanceof MetricDatum) {
        return Arrays.asList((MetricDatum) body);
    }

    MetricDatum metricDatum = new MetricDatum()
            .withMetricName(determineName(exchange))
            .withValue(determineValue(exchange))
            .withUnit(determineUnit(exchange))
            .withTimestamp(determineTimestamp(exchange));
    setDimension(metricDatum, exchange);
    return Arrays.asList(metricDatum);
}

private void setDimension(MetricDatum metricDatum, Exchange exchange) {
    String name = exchange.getIn().getHeader(CwConstants.METRIC_DIMENSION_NAME, String.class);
    String value = exchange.getIn().getHeader(CwConstants.METRIC_DIMENSION_VALUE, String.class);
    if (name != null && value != null) {
        metricDatum.withDimensions(new Dimension().withName(name).withValue(value));
    } else {
        Map<String, String> dimensions = exchange.getIn().getHeader(CwConstants.METRIC_DIMENSIONS, Map.class);
        if (dimensions != null) {
            Collection<Dimension> dimensionCollection = new ArrayList<Dimension>();
            for (Map.Entry<String, String> dimensionEntry : dimensions.entrySet()) {
                Dimension dimension = new Dimension().withName(dimensionEntry.getKey()).withValue(dimensionEntry.getValue());
                dimensionCollection.add(dimension);
            }
            metricDatum.withDimensions(dimensionCollection);
        }
    }
}

@Test
public void sendManuallyCreatedMetric() throws Exception {
    template.send("direct:start", ExchangePattern.InOnly, new Processor() {
        public void process(Exchange exchange) throws Exception {
            MetricDatum metricDatum = new MetricDatum()
                    .withMetricName("errorCount")
                    .withValue(Double.valueOf(0));
            exchange.getIn().setBody(metricDatum);
        }
    });


    ArgumentCaptor<PutMetricDataRequest> argument = ArgumentCaptor.forClass(PutMetricDataRequest.class);
    verify(cloudWatchClient).putMetricData(argument.capture());

    assertEquals("errorCount", argument.getValue().getMetricData().get(0).getMetricName());
    assertEquals(Double.valueOf(0), argument.getValue().getMetricData().get(0).getValue());
}

@Override
public void send() throws IOException {
    final List<MetricDatum> datumList;
    synchronized (buffer) {
        datumList = new ArrayList<MetricDatum>(buffer);
        buffer.clear();
    }

    // note: Each PutMetricData request is limited to 40 KB in size for HTTP POST requests.
    final PutMetricDataRequest request = new PutMetricDataRequest()
            .withNamespace(cloudWatchNamespace).withMetricData(datumList);
    try {
        awsCloudWatch.putMetricData(request);
    } catch (final Exception e) {
        // pas catch (AmazonCloudWatchException) sinon ClassNotFoundException dans Jenkins par ex
        throw new IOException("Error connecting to AWS CloudWatch", e);
    }
}

@Override
public void run() {
    try {
        List<MetricDatum> metricData = new ArrayList<>();

        Date now = new Date();
        metricData.addAll(heartbeats.entrySet().stream().map(entry -> new MetricDatum().withMetricName("Heartbeats")
                                                                                       .withDimensions(new Dimension().withName("Client").withValue(entry.getKey()))
                                                                                       .withTimestamp(now)
                                                                                       .withUnit(StandardUnit.Count)
                                                                                       .withValue(entry.getValue().doubleValue())).collect(Collectors.toList()));
        heartbeats.clear();

            for (List<MetricDatum> chunk :partitionList(metricData, 20)) {
                awsClient.putMetricData(new PutMetricDataRequest().withNamespace(namespace).withMetricData(chunk));
            }

    } catch (Throwable e) {
        log.error("Failed to publish CloudWatch metrics: {}", e.toString());
    }
}

/**
 * @return a string representation of this object
 */
@Override
public String toString() {
    StringBuilder sb = new StringBuilder("{ElapsedTimeAggregator: {");
    String namespace = putMetricDataRequest.getNamespace();
    sb.append("region: ").append(region).append(", ");
    sb.append("namespace: ").append(namespace).append(", ");
    sb.append("metrics: [");
    String msep = "";
    for (MetricDatum metricDatum : putMetricDataRequest.getMetricData()) {
        sb.append(msep).append("{metricName: ").append(metricDatum.getMetricName()).append(", ");
        sb.append("dimensions: [");
        String dsep = "";
        for (Dimension dimension : metricDatum.getDimensions()) {
            sb.append(dsep).append(dimension);
            dsep = ", ";
        }
        sb.append("]}");
        msep = ", ";
    }
    sb.append("]}}");
    return sb.toString();
}

protected void sendBatch(Map<String, Collection<MetricDatum>> datums) {
    for (final Map.Entry<String, Collection<MetricDatum>> e : datums.entrySet()) {
        for (final List<MetricDatum> batch : Lists.partition(Lists.newLinkedList(e.getValue()), BATCH_SIZE)) {
            cloudWatch.putMetricDataAsync(new PutMetricDataRequest().withNamespace(e.getKey())
                                                                    .withMetricData(batch),
                                          new AsyncHandler<PutMetricDataRequest, Void>() {

                                            @Override
                                            public void onError(Exception exception) {
                                                LOG.error("PutMetricData failed", exception);
                                                LOG.info("Requeueing metric data.");
                                                queuedDatums.putAll(e.getKey(), batch);
                                            }

                                            @Override
                                            public void onSuccess(PutMetricDataRequest request, Void result) {
                                                LOG.info("Successfully put " + request.getMetricData().size() +
                                                                        " datums for namespace " + request.getNamespace());
                                                LOG.debug("Request", request);
                                            }

            });
        }
    }
}

@Override
protected void runOnce() throws Exception {
    ClickEvent event = inputQueue.take();
    String partitionKey = event.getSessionId();
    ByteBuffer data = ByteBuffer.wrap(
            event.getPayload().getBytes("UTF-8"));
    recordsPut.getAndIncrement();

    PutRecordResult res = kinesis.putRecord(
            STREAM_NAME, data, partitionKey);

    MetricDatum d = new MetricDatum()
        .withDimensions(
            new Dimension().withName("StreamName").withValue(STREAM_NAME),
            new Dimension().withName("ShardId").withValue(res.getShardId()),
            new Dimension().withName("Host").withValue(
                    InetAddress.getLocalHost().toString()))
        .withValue(1.0)
        .withMetricName("RecordsPut");
    cw.putMetricData(new PutMetricDataRequest()
        .withMetricData(d)
        .withNamespace("MySampleProducer"));
}

void reportGauge(Map.Entry<String, Gauge> gaugeEntry, String typeDimValue, List<MetricDatum> data) {
    Gauge gauge = gaugeEntry.getValue();

    Object valueObj = gauge.getValue();
    if (valueObj == null) {
        return;
    }

    String valueStr = valueObj.toString();
    if (NumberUtils.isNumber(valueStr)) {
        final Number value = NumberUtils.createNumber(valueStr);

        DemuxedKey key = new DemuxedKey(appendGlobalDimensions(gaugeEntry.getKey()));
        Iterables.addAll(data, key.newDatums(typeDimName, typeDimValue, new Function<MetricDatum, MetricDatum>() {
            @Override
            public MetricDatum apply(MetricDatum datum) {
                return datum.withValue(value.doubleValue());
            }
        }));
    }
}

void reportCounter(Map.Entry<String, ? extends Counting> entry, String typeDimValue, List<MetricDatum> data) {
    Counting metric = entry.getValue();
    final long diff = diffLast(metric);
    if (diff == 0) {
        // Don't submit metrics that have not changed. No reason to keep these alive. Also saves on CloudWatch
        // costs.
        return;
    }

    DemuxedKey key = new DemuxedKey(appendGlobalDimensions(entry.getKey()));
    Iterables.addAll(data, key.newDatums(typeDimName, typeDimValue, new Function<MetricDatum, MetricDatum>() {
        @Override
        public MetricDatum apply(MetricDatum datum) {
            return datum.withValue((double) diff).withUnit(StandardUnit.Count);
        }
    }));
}

/**
 * @param rescale the submitted sum by this multiplier. 1.0 is the identity (no rescale).
 */
void reportSampling(Map.Entry<String, ? extends Sampling> entry, String typeDimValue, double rescale, List<MetricDatum> data) {
    Sampling metric = entry.getValue();
    Snapshot snapshot = metric.getSnapshot();
    double scaledSum = sum(snapshot.getValues()) * rescale;
    final StatisticSet statisticSet = new StatisticSet()
            .withSum(scaledSum)
            .withSampleCount((double) snapshot.size())
            .withMinimum((double) snapshot.getMin() * rescale)
            .withMaximum((double) snapshot.getMax() * rescale);

    DemuxedKey key = new DemuxedKey(appendGlobalDimensions(entry.getKey()));
    Iterables.addAll(data, key.newDatums(typeDimName, typeDimValue, new Function<MetricDatum, MetricDatum>() {
        @Override
        public MetricDatum apply(MetricDatum datum) {
            return datum.withStatisticValues(statisticSet);
        }
    }));
}

public MetricDatum getDatum() {
    MetricDatum md = new MetricDatum();
    md.setMetricName(name);
    md.setUnit(unit);
    md.setDimensions(dimensions);
    return md;
}

@Test
public void record_and_shutdown() throws Exception {
    final DimensionMapper mapper = new DimensionMapper.Builder()
            .addGlobalDimension(ContextData.ID)
            .build();

    final String namespace = "spacename";

    final String id = UUID.randomUUID().toString();
    final TypedMap data = ContextData.withId(id).build();
    final double time = 123.45;

    final AmazonCloudWatch client = mock(AmazonCloudWatch.class);

    final CloudWatchRecorder recorder = new CloudWatchRecorder(client, namespace, 60, 120, mapper);

    final MetricRecorder.Context context = recorder.context(data);
    context.record(M_TIME, time, Unit.MILLISECOND, Instant.now());
    context.close();

    // shutdown right away, before first scheduled publish
    recorder.shutdown();

    final List<MetricDatum> expected = new ArrayList<>(1);
    expected.add(makeDatum(id, M_TIME.toString(), time, time, time, 1, StandardUnit.Milliseconds));

    verify(client).putMetricData(argThat(new RequestMatcher(namespace, expected)));
}

@Test
public void single() throws Exception {
    final DimensionMapper mapper = new DimensionMapper.Builder()
            .addGlobalDimension(ContextData.ID)
            .build();

    final Metric name = Metric.define("SomeMetric");
    final double value = 3.14;
    final StandardUnit unit = StandardUnit.Terabits;

    final TypedMap context = ContextData.withId(UUID.randomUUID().toString()).build();

    MetricDataAggregator aggregator = new MetricDataAggregator(mapper);
    aggregator.add(context, name, value, unit);

    List<MetricDatum> ags = aggregator.flush();

    assertEquals("One metric datum should aggregate to one entry", 1, ags.size());
    assertEquals("Metric datum has wrong name", name.toString(), ags.get(0).getMetricName());
    assertEquals("Metric datum has wrong unit", unit.toString(), ags.get(0).getUnit());

    StatisticSet stats = ags.get(0).getStatisticValues();
    assertEquals("Metric datum has wrong stats value", Double.valueOf(value), stats.getSum());
    assertEquals("Metric datum has wrong stats value", Double.valueOf(value), stats.getMinimum());
    assertEquals("Metric datum has wrong stats value", Double.valueOf(value), stats.getMaximum());
    assertEquals("Metric datum has wrong stats count", Double.valueOf(1), stats.getSampleCount());

    assertTrue("Flush with no data was non-empty", aggregator.flush().isEmpty());
}

/**
 * Send a datum object to this metric.
 *
 * @param value the measurement to record
 */
public void sendValue(double value) {
    MetricDatum datum = new MetricDatum();
    datum.setMetricName(name);
    datum.setUnit(unit);
    datum.setValue(value);

    PutMetricDataRequest request = new PutMetricDataRequest();
    request.setNamespace(namespace);
    request.withMetricData(datum);

    logger.info(String.format("logging %.2f to cloudwatch metric %s/%s", value, namespace, name));
    client.putMetricData(request);
}

private void postToCloudwatch(String namespace, List<CloudwatchStatistic> statistics) {
    List<MetricDatum> metricDatumList = statistics.stream()
            .map(CloudwatchStatistic::toMetricsDatum)
            .collect(toList());

    PutMetricDataRequest request = new PutMetricDataRequest();
    request.setNamespace(namespace);
    request.setMetricData(metricDatumList);

    amazonCloudWatch.putMetricData(request);
}

public MetricDatum toMetricsDatum() {
    MetricDatum metricDatum = new MetricDatum();
    metricDatum.setMetricName(key);
    metricDatum.setTimestamp(new Date());
    metricDatum.setUnit(unit);
    metricDatum.setValue(value);
    return metricDatum;
}

private void processCounter(final String metricName, final Counting counter, final List<MetricDatum> metricData) {
    long currentCount = counter.getCount();
    Long lastCount = lastPolledCounts.get(counter);
    lastPolledCounts.put(counter, currentCount);

    if (lastCount == null) {
        lastCount = 0L;
    }

    // Only submit metrics that have changed - let's save some money!
    final long delta = currentCount - lastCount;
    stageMetricDatum(true, metricName, delta, StandardUnit.Count, DIMENSION_COUNT, metricData);
}

@Test
public void shouldReportExpectedCounterValue() throws Exception {
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    reporterBuilder.build().report();

    final MetricDatum metricDatum = firstMetricDatumFromCapturedRequest();

    assertThat(metricDatum.getValue()).isWithin(1.0);
    assertThat(metricDatum.getUnit()).isEqualTo(Count.toString());
}

@Test
public void shouldNotReportUnchangedCounterValue() throws Exception {
    metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
    final CloudWatchReporter cloudWatchReporter = reporterBuilder.build();

    cloudWatchReporter.report();
    MetricDatum metricDatum = firstMetricDatumFromCapturedRequest();
    assertThat(metricDatum.getValue().intValue()).isEqualTo(1);
    metricDataRequestCaptor.getAllValues().clear();

    cloudWatchReporter.report();

    verify(mockAmazonCloudWatchAsyncClient, times(1)).putMetricDataAsync(any(PutMetricDataRequest.class));
}

@Test
public void shouldReportArithmeticMeanWithoutConversionWhenReportingHistogram() throws Exception {
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(1);
    reporterBuilder.withArithmeticMean().build().report();

    final MetricDatum metricData = metricDatumByDimensionFromCapturedRequest(DIMENSION_SNAPSHOT_MEAN);

    assertThat(metricData.getValue().intValue()).isEqualTo(1);
    assertThat(metricData.getUnit()).isEqualTo(None.toString());
}

@Test
public void shouldReportStdDevWithoutConversionWhenReportingHistogram() throws Exception {
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(1);
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(2);
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(3);
    metricRegistry.histogram(CloudWatchReporterTest.ARBITRARY_HISTOGRAM_NAME).update(30);
    reporterBuilder.withStdDev().build().report();

    final MetricDatum metricData = metricDatumByDimensionFromCapturedRequest(DIMENSION_SNAPSHOT_STD_DEV);

    assertThat(metricData.getValue().intValue()).isEqualTo(12);
    assertThat(metricData.getUnit()).isEqualTo(None.toString());
}

@Test
public void shouldReportHistogramSubsequentSnapshotValues_SumMaxMinValues() throws Exception {
    CloudWatchReporter reporter = reporterBuilder.withStatisticSet().build();

    final Histogram slidingWindowHistogram = new Histogram(new SlidingWindowReservoir(4));
    metricRegistry.register("SlidingWindowHistogram", slidingWindowHistogram);

    slidingWindowHistogram.update(1);
    slidingWindowHistogram.update(2);
    slidingWindowHistogram.update(30);
    reporter.report();

    final MetricDatum metricData = metricDatumByDimensionFromCapturedRequest(DIMENSION_SNAPSHOT_SUMMARY);

    assertThat(metricData.getStatisticValues().getMaximum().intValue()).isEqualTo(30);
    assertThat(metricData.getStatisticValues().getMinimum().intValue()).isEqualTo(1);
    assertThat(metricData.getStatisticValues().getSampleCount().intValue()).isEqualTo(3);
    assertThat(metricData.getStatisticValues().getSum().intValue()).isEqualTo(33);
    assertThat(metricData.getUnit()).isEqualTo(None.toString());

    slidingWindowHistogram.update(4);
    slidingWindowHistogram.update(100);
    slidingWindowHistogram.update(5);
    slidingWindowHistogram.update(6);
    reporter.report();

    final MetricDatum secondMetricData = metricDatumByDimensionFromCapturedRequest(DIMENSION_SNAPSHOT_SUMMARY);

    assertThat(secondMetricData.getStatisticValues().getMaximum().intValue()).isEqualTo(100);
    assertThat(secondMetricData.getStatisticValues().getMinimum().intValue()).isEqualTo(4);
    assertThat(secondMetricData.getStatisticValues().getSampleCount().intValue()).isEqualTo(4);
    assertThat(secondMetricData.getStatisticValues().getSum().intValue()).isEqualTo(115);
    assertThat(secondMetricData.getUnit()).isEqualTo(None.toString());

}

private List<Dimension> allDimensionsFromCapturedRequest() {
    final PutMetricDataRequest putMetricDataRequest = metricDataRequestCaptor.getValue();
    final List<MetricDatum> metricData = putMetricDataRequest.getMetricData();
    final List<Dimension> all = new LinkedList<>();
    for (final MetricDatum metricDatum : metricData) {
        all.addAll(metricDatum.getDimensions());
    }
    return all;
}

public static void main(String[] args) {

        final String USAGE =
            "To run this example, supply a data point:\n" +
            "Ex: PutMetricData <data_point>\n";

        if (args.length != 1) {
            System.out.println(USAGE);
            System.exit(1);
        }

        Double data_point = Double.parseDouble(args[0]);

        final AmazonCloudWatch cw =
            AmazonCloudWatchClientBuilder.defaultClient();

        Dimension dimension = new Dimension()
            .withName("UNIQUE_PAGES")
            .withValue("URLS");

        MetricDatum datum = new MetricDatum()
            .withMetricName("PAGES_VISITED")
            .withUnit(StandardUnit.None)
            .withValue(data_point)
            .withDimensions(dimension);

        PutMetricDataRequest request = new PutMetricDataRequest()
            .withNamespace("SITE/TRAFFIC")
            .withMetricData(datum);

        PutMetricDataResult response = cw.putMetricData(request);

        System.out.printf("Successfully put data point %f", data_point);
    }

private void reportTimer(String key, Collection<MetricDatum> data, Map.Entry<String, Timer> met) {
    Timer timer = met.getValue();
    Snapshot snapshot = timer.getSnapshot();
    if (reportAggregates) {
        reportAggregate(key, data, "count", null, timer.getCount());
        reportAggregate(key, data, "rate", "1minute", timer.getOneMinuteRate());
        reportAggregate(key, data, "rate", "5minute", timer.getFiveMinuteRate());
        reportAggregate(key, data, "rate", "15minute", timer.getFifteenMinuteRate());
        reportAggregate(key, data, "rate", "mean", timer.getMeanRate());
        reportSnapshot(data, snapshot, key);
    } else {
        // if no data, don't bother Amazon with it.
        if (snapshot.size() == 0) {
            return;
        }
        double sum = 0;
        for (double val : snapshot.getValues()) {
            sum += val;
        }
        // Metrics works in Nanoseconds, which is not one of Amazon's favorites.
        double max = (double) TimeUnit.NANOSECONDS.toMicros(snapshot.getMax());
        double min = (double) TimeUnit.NANOSECONDS.toMicros(snapshot.getMin());
        double sumMicros = TimeUnit.NANOSECONDS.toMicros((long) sum);
        StatisticSet stats = new StatisticSet()
                .withMaximum(max)
                .withMinimum(min)
                .withSum(sumMicros)
                .withSampleCount((double) snapshot.getValues().length);
        if (LOG.isDebugEnabled()) {
            LOG.debug("timer {}: {}", met.getKey(), stats);
        }
        data.add(new MetricDatum().withMetricName(met.getKey())
                .withDimensions(dimensions)
                .withStatisticValues(stats)
                .withUnit(StandardUnit.Microseconds));
    }
}

private void reportAggregate(String key, Collection<MetricDatum> data, String valDimName,
                             String valDimValue, double value) {
    Collection<Dimension> fullDimensions = new ArrayList<>();
    fullDimensions.addAll(dimensions);
    fullDimensions.add(new Dimension().withName(valDimName).withValue(valDimValue));
    data.add(new MetricDatum().withMetricName(key)
            .withDimensions(fullDimensions)
            .withValue(value));

}