@Test public void testAddNewRecordTimeout() throws InterruptedException { long timeBucketIntervalInSec = 60; int reportFreqMsgCount = 10; short reportFreqIntervalSec = 2; AuditAggregator agg = new AuditAggregator(timeBucketIntervalInSec, reportFreqMsgCount, reportFreqIntervalSec); int partition = 0; long offset = 1024; long nowInMs = System.currentTimeMillis(); JSONObject record = createRecordMock("mytopic", 2, 5, 95, nowInMs); EasyMock.replay(record); assertFalse(agg.addRecord(AUDIT_TOPIC, partition, offset, record)); Thread.sleep(3000); assertTrue(agg.addRecord(AUDIT_TOPIC, partition, offset, record)); EasyMock.verify(record); }
public boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record) { String topicName = record.getString(AuditMsgField.TOPICNAME.getName()); String hostName = record.getString(AuditMsgField.HOSTNAME.getName()); String tier = record.getString(AuditMsgField.TIER.getName()); String dc = record.getString(AuditMsgField.DATACENTER.getName()); if (combineMetricsAmongHosts) { hostName = "combined"; } String key = String.format(KEY_FORMAT, topicName, hostName, tier, dc); double timestampInSec = record.getDoubleValue(AuditMsgField.TIME_BUCKET_START.getName()); if (timestampInSec > MIN_TIMESTAMP_MS) { timestampInSec /= 1000; } if (timestampInSec <= 0 || timestampInSec >= (timestampSkewSec + System.currentTimeMillis() / 1000.0)) { logger.warn(String.format("Got very skewed timestamp from %s as %f", topicName, timestampInSec)); INVALID_TIMESTAMP_COUNTER.mark(); return false; } long totalCount = record.getLongValue(AuditMsgField.METRICS_COUNT.getName()); double meanLatency = record.getDoubleValue(AuditMsgField.METRICS_MEAN_LATENCY.getName()); double p95Latency = record.getDoubleValue(AuditMsgField.METRICS_P95_LATENCY.getName()); long totalBytes = record.getLongValue(AuditMsgField.METRICS_TOTAL_BYTES.getName()); long noTimeCount = record.getLongValue(AuditMsgField.METRICS_COUNT_NOTIME.getName()); long malformedCount = record.getLongValue(AuditMsgField.METRICS_COUNT_MALFORMED.getName()); double p99Latency = record.getDoubleValue(AuditMsgField.METRICS_P99_LATENCY.getName()); double maxLatency = record.getDoubleValue(AuditMsgField.METRICS_MAX_LATENCY.getName()); long bucketTimeInSec = (long) (timestampInSec - timestampInSec % timeBucketIntervalInSec); Map<String, TimeBucket> timeBuckets = buffer.get(bucketTimeInSec); if (timeBuckets == null) { timeBuckets = new HashMap<>(); buffer.put(bucketTimeInSec, timeBuckets); } TimeBucket timeBucket = timeBuckets.get(key); if (timeBucket == null) { timeBucket = new TimeBucket(bucketTimeInSec, bucketTimeInSec + timeBucketIntervalInSec, topicName, hostName, tier, dc); timeBuckets.put(key, timeBucket); } timeBucket.addNewRecord(totalBytes, totalCount, noTimeCount, malformedCount, meanLatency, p95Latency, p99Latency, maxLatency); updateOffset(sourceTopic, partition, offset); long bucketedMsgCount = timeBucket.count; if (bucketedMsgCount >= reportFreqMsgCount) { logger.info(String.format("FreqMsgCount is reached for freqMsgCount=%d, bucketedMsgCount=%d", reportFreqMsgCount, bucketedMsgCount)); } long currentTimeInMs = System.currentTimeMillis(); if (currentTimeInMs > nextReportTimeInMs) { logger.info(String.format("FreqTimeInterval is reached currentTimeInMs=%d, nextTimeInMs=%d", currentTimeInMs, nextReportTimeInMs)); } return (bucketedMsgCount >= reportFreqMsgCount || currentTimeInMs > nextReportTimeInMs); }
AuditAggregator { public boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record) { String topicName = record.getString(AuditMsgField.TOPICNAME.getName()); String hostName = record.getString(AuditMsgField.HOSTNAME.getName()); String tier = record.getString(AuditMsgField.TIER.getName()); String dc = record.getString(AuditMsgField.DATACENTER.getName()); if (combineMetricsAmongHosts) { hostName = "combined"; } String key = String.format(KEY_FORMAT, topicName, hostName, tier, dc); double timestampInSec = record.getDoubleValue(AuditMsgField.TIME_BUCKET_START.getName()); if (timestampInSec > MIN_TIMESTAMP_MS) { timestampInSec /= 1000; } if (timestampInSec <= 0 || timestampInSec >= (timestampSkewSec + System.currentTimeMillis() / 1000.0)) { logger.warn(String.format("Got very skewed timestamp from %s as %f", topicName, timestampInSec)); INVALID_TIMESTAMP_COUNTER.mark(); return false; } long totalCount = record.getLongValue(AuditMsgField.METRICS_COUNT.getName()); double meanLatency = record.getDoubleValue(AuditMsgField.METRICS_MEAN_LATENCY.getName()); double p95Latency = record.getDoubleValue(AuditMsgField.METRICS_P95_LATENCY.getName()); long totalBytes = record.getLongValue(AuditMsgField.METRICS_TOTAL_BYTES.getName()); long noTimeCount = record.getLongValue(AuditMsgField.METRICS_COUNT_NOTIME.getName()); long malformedCount = record.getLongValue(AuditMsgField.METRICS_COUNT_MALFORMED.getName()); double p99Latency = record.getDoubleValue(AuditMsgField.METRICS_P99_LATENCY.getName()); double maxLatency = record.getDoubleValue(AuditMsgField.METRICS_MAX_LATENCY.getName()); long bucketTimeInSec = (long) (timestampInSec - timestampInSec % timeBucketIntervalInSec); Map<String, TimeBucket> timeBuckets = buffer.get(bucketTimeInSec); if (timeBuckets == null) { timeBuckets = new HashMap<>(); buffer.put(bucketTimeInSec, timeBuckets); } TimeBucket timeBucket = timeBuckets.get(key); if (timeBucket == null) { timeBucket = new TimeBucket(bucketTimeInSec, bucketTimeInSec + timeBucketIntervalInSec, topicName, hostName, tier, dc); timeBuckets.put(key, timeBucket); } timeBucket.addNewRecord(totalBytes, totalCount, noTimeCount, malformedCount, meanLatency, p95Latency, p99Latency, maxLatency); updateOffset(sourceTopic, partition, offset); long bucketedMsgCount = timeBucket.count; if (bucketedMsgCount >= reportFreqMsgCount) { logger.info(String.format("FreqMsgCount is reached for freqMsgCount=%d, bucketedMsgCount=%d", reportFreqMsgCount, bucketedMsgCount)); } long currentTimeInMs = System.currentTimeMillis(); if (currentTimeInMs > nextReportTimeInMs) { logger.info(String.format("FreqTimeInterval is reached currentTimeInMs=%d, nextTimeInMs=%d", currentTimeInMs, nextReportTimeInMs)); } return (bucketedMsgCount >= reportFreqMsgCount || currentTimeInMs > nextReportTimeInMs); } }
AuditAggregator { public boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record) { String topicName = record.getString(AuditMsgField.TOPICNAME.getName()); String hostName = record.getString(AuditMsgField.HOSTNAME.getName()); String tier = record.getString(AuditMsgField.TIER.getName()); String dc = record.getString(AuditMsgField.DATACENTER.getName()); if (combineMetricsAmongHosts) { hostName = "combined"; } String key = String.format(KEY_FORMAT, topicName, hostName, tier, dc); double timestampInSec = record.getDoubleValue(AuditMsgField.TIME_BUCKET_START.getName()); if (timestampInSec > MIN_TIMESTAMP_MS) { timestampInSec /= 1000; } if (timestampInSec <= 0 || timestampInSec >= (timestampSkewSec + System.currentTimeMillis() / 1000.0)) { logger.warn(String.format("Got very skewed timestamp from %s as %f", topicName, timestampInSec)); INVALID_TIMESTAMP_COUNTER.mark(); return false; } long totalCount = record.getLongValue(AuditMsgField.METRICS_COUNT.getName()); double meanLatency = record.getDoubleValue(AuditMsgField.METRICS_MEAN_LATENCY.getName()); double p95Latency = record.getDoubleValue(AuditMsgField.METRICS_P95_LATENCY.getName()); long totalBytes = record.getLongValue(AuditMsgField.METRICS_TOTAL_BYTES.getName()); long noTimeCount = record.getLongValue(AuditMsgField.METRICS_COUNT_NOTIME.getName()); long malformedCount = record.getLongValue(AuditMsgField.METRICS_COUNT_MALFORMED.getName()); double p99Latency = record.getDoubleValue(AuditMsgField.METRICS_P99_LATENCY.getName()); double maxLatency = record.getDoubleValue(AuditMsgField.METRICS_MAX_LATENCY.getName()); long bucketTimeInSec = (long) (timestampInSec - timestampInSec % timeBucketIntervalInSec); Map<String, TimeBucket> timeBuckets = buffer.get(bucketTimeInSec); if (timeBuckets == null) { timeBuckets = new HashMap<>(); buffer.put(bucketTimeInSec, timeBuckets); } TimeBucket timeBucket = timeBuckets.get(key); if (timeBucket == null) { timeBucket = new TimeBucket(bucketTimeInSec, bucketTimeInSec + timeBucketIntervalInSec, topicName, hostName, tier, dc); timeBuckets.put(key, timeBucket); } timeBucket.addNewRecord(totalBytes, totalCount, noTimeCount, malformedCount, meanLatency, p95Latency, p99Latency, maxLatency); updateOffset(sourceTopic, partition, offset); long bucketedMsgCount = timeBucket.count; if (bucketedMsgCount >= reportFreqMsgCount) { logger.info(String.format("FreqMsgCount is reached for freqMsgCount=%d, bucketedMsgCount=%d", reportFreqMsgCount, bucketedMsgCount)); } long currentTimeInMs = System.currentTimeMillis(); if (currentTimeInMs > nextReportTimeInMs) { logger.info(String.format("FreqTimeInterval is reached currentTimeInMs=%d, nextTimeInMs=%d", currentTimeInMs, nextReportTimeInMs)); } return (bucketedMsgCount >= reportFreqMsgCount || currentTimeInMs > nextReportTimeInMs); } AuditAggregator(long timeBucketIntervalInSec, int reportFreqMsgCount, int reportFreqIntervalSec);  AuditAggregator(long timeBucketIntervalInSec, int reportFreqMsgCount, int reportFreqIntervalSec,
      boolean combineMetricsAmongHosts); }
AuditAggregator { public boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record) { String topicName = record.getString(AuditMsgField.TOPICNAME.getName()); String hostName = record.getString(AuditMsgField.HOSTNAME.getName()); String tier = record.getString(AuditMsgField.TIER.getName()); String dc = record.getString(AuditMsgField.DATACENTER.getName()); if (combineMetricsAmongHosts) { hostName = "combined"; } String key = String.format(KEY_FORMAT, topicName, hostName, tier, dc); double timestampInSec = record.getDoubleValue(AuditMsgField.TIME_BUCKET_START.getName()); if (timestampInSec > MIN_TIMESTAMP_MS) { timestampInSec /= 1000; } if (timestampInSec <= 0 || timestampInSec >= (timestampSkewSec + System.currentTimeMillis() / 1000.0)) { logger.warn(String.format("Got very skewed timestamp from %s as %f", topicName, timestampInSec)); INVALID_TIMESTAMP_COUNTER.mark(); return false; } long totalCount = record.getLongValue(AuditMsgField.METRICS_COUNT.getName()); double meanLatency = record.getDoubleValue(AuditMsgField.METRICS_MEAN_LATENCY.getName()); double p95Latency = record.getDoubleValue(AuditMsgField.METRICS_P95_LATENCY.getName()); long totalBytes = record.getLongValue(AuditMsgField.METRICS_TOTAL_BYTES.getName()); long noTimeCount = record.getLongValue(AuditMsgField.METRICS_COUNT_NOTIME.getName()); long malformedCount = record.getLongValue(AuditMsgField.METRICS_COUNT_MALFORMED.getName()); double p99Latency = record.getDoubleValue(AuditMsgField.METRICS_P99_LATENCY.getName()); double maxLatency = record.getDoubleValue(AuditMsgField.METRICS_MAX_LATENCY.getName()); long bucketTimeInSec = (long) (timestampInSec - timestampInSec % timeBucketIntervalInSec); Map<String, TimeBucket> timeBuckets = buffer.get(bucketTimeInSec); if (timeBuckets == null) { timeBuckets = new HashMap<>(); buffer.put(bucketTimeInSec, timeBuckets); } TimeBucket timeBucket = timeBuckets.get(key); if (timeBucket == null) { timeBucket = new TimeBucket(bucketTimeInSec, bucketTimeInSec + timeBucketIntervalInSec, topicName, hostName, tier, dc); timeBuckets.put(key, timeBucket); } timeBucket.addNewRecord(totalBytes, totalCount, noTimeCount, malformedCount, meanLatency, p95Latency, p99Latency, maxLatency); updateOffset(sourceTopic, partition, offset); long bucketedMsgCount = timeBucket.count; if (bucketedMsgCount >= reportFreqMsgCount) { logger.info(String.format("FreqMsgCount is reached for freqMsgCount=%d, bucketedMsgCount=%d", reportFreqMsgCount, bucketedMsgCount)); } long currentTimeInMs = System.currentTimeMillis(); if (currentTimeInMs > nextReportTimeInMs) { logger.info(String.format("FreqTimeInterval is reached currentTimeInMs=%d, nextTimeInMs=%d", currentTimeInMs, nextReportTimeInMs)); } return (bucketedMsgCount >= reportFreqMsgCount || currentTimeInMs > nextReportTimeInMs); } AuditAggregator(long timeBucketIntervalInSec, int reportFreqMsgCount, int reportFreqIntervalSec);  AuditAggregator(long timeBucketIntervalInSec, int reportFreqMsgCount, int reportFreqIntervalSec,
      boolean combineMetricsAmongHosts); boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record); Map<Long/* begin_time */, Map<String, TimeBucket>> getAndResetBuffer(); Map<String, Map<Integer, Long>> getOffsets(); }
AuditAggregator { public boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record) { String topicName = record.getString(AuditMsgField.TOPICNAME.getName()); String hostName = record.getString(AuditMsgField.HOSTNAME.getName()); String tier = record.getString(AuditMsgField.TIER.getName()); String dc = record.getString(AuditMsgField.DATACENTER.getName()); if (combineMetricsAmongHosts) { hostName = "combined"; } String key = String.format(KEY_FORMAT, topicName, hostName, tier, dc); double timestampInSec = record.getDoubleValue(AuditMsgField.TIME_BUCKET_START.getName()); if (timestampInSec > MIN_TIMESTAMP_MS) { timestampInSec /= 1000; } if (timestampInSec <= 0 || timestampInSec >= (timestampSkewSec + System.currentTimeMillis() / 1000.0)) { logger.warn(String.format("Got very skewed timestamp from %s as %f", topicName, timestampInSec)); INVALID_TIMESTAMP_COUNTER.mark(); return false; } long totalCount = record.getLongValue(AuditMsgField.METRICS_COUNT.getName()); double meanLatency = record.getDoubleValue(AuditMsgField.METRICS_MEAN_LATENCY.getName()); double p95Latency = record.getDoubleValue(AuditMsgField.METRICS_P95_LATENCY.getName()); long totalBytes = record.getLongValue(AuditMsgField.METRICS_TOTAL_BYTES.getName()); long noTimeCount = record.getLongValue(AuditMsgField.METRICS_COUNT_NOTIME.getName()); long malformedCount = record.getLongValue(AuditMsgField.METRICS_COUNT_MALFORMED.getName()); double p99Latency = record.getDoubleValue(AuditMsgField.METRICS_P99_LATENCY.getName()); double maxLatency = record.getDoubleValue(AuditMsgField.METRICS_MAX_LATENCY.getName()); long bucketTimeInSec = (long) (timestampInSec - timestampInSec % timeBucketIntervalInSec); Map<String, TimeBucket> timeBuckets = buffer.get(bucketTimeInSec); if (timeBuckets == null) { timeBuckets = new HashMap<>(); buffer.put(bucketTimeInSec, timeBuckets); } TimeBucket timeBucket = timeBuckets.get(key); if (timeBucket == null) { timeBucket = new TimeBucket(bucketTimeInSec, bucketTimeInSec + timeBucketIntervalInSec, topicName, hostName, tier, dc); timeBuckets.put(key, timeBucket); } timeBucket.addNewRecord(totalBytes, totalCount, noTimeCount, malformedCount, meanLatency, p95Latency, p99Latency, maxLatency); updateOffset(sourceTopic, partition, offset); long bucketedMsgCount = timeBucket.count; if (bucketedMsgCount >= reportFreqMsgCount) { logger.info(String.format("FreqMsgCount is reached for freqMsgCount=%d, bucketedMsgCount=%d", reportFreqMsgCount, bucketedMsgCount)); } long currentTimeInMs = System.currentTimeMillis(); if (currentTimeInMs > nextReportTimeInMs) { logger.info(String.format("FreqTimeInterval is reached currentTimeInMs=%d, nextTimeInMs=%d", currentTimeInMs, nextReportTimeInMs)); } return (bucketedMsgCount >= reportFreqMsgCount || currentTimeInMs > nextReportTimeInMs); } AuditAggregator(long timeBucketIntervalInSec, int reportFreqMsgCount, int reportFreqIntervalSec);  AuditAggregator(long timeBucketIntervalInSec, int reportFreqMsgCount, int reportFreqIntervalSec,
      boolean combineMetricsAmongHosts); boolean addRecord(String sourceTopic, int partition, long offset, JSONObject record); Map<Long/* begin_time */, Map<String, TimeBucket>> getAndResetBuffer(); Map<String, Map<Integer, Long>> getOffsets(); static final String KEY_FORMAT; }