Adds an exported function to flush internal reader (#1248)
diff --git a/metric/metricexport/reader.go b/metric/metricexport/reader.go
index b920bac..8a09d0f 100644
--- a/metric/metricexport/reader.go
+++ b/metric/metricexport/reader.go
@@ -130,7 +130,7 @@
reportingInterval = ir.ReportingInterval
}
- if ir.done != nil {
+ if ir.quit != nil {
return errAlreadyStarted
}
ir.timer = time.NewTicker(reportingInterval)
@@ -172,6 +172,19 @@
ir.quit = nil
}
+// Flush flushes the metrics if IntervalReader is stopped, otherwise no-op.
+func (ir *IntervalReader) Flush() {
+ ir.mu.Lock()
+ defer ir.mu.Unlock()
+
+ // No-op if IntervalReader is not stopped
+ if ir.quit != nil {
+ return
+ }
+
+ ir.reader.ReadAndExport(ir.exporter)
+}
+
// ReadAndExport reads metrics from all producer registered with
// producer manager and then exports them using provided exporter.
func (r *Reader) ReadAndExport(exporter Exporter) {
diff --git a/metric/metricexport/reader_test.go b/metric/metricexport/reader_test.go
index 33313dc..61c915b 100644
--- a/metric/metricexport/reader_test.go
+++ b/metric/metricexport/reader_test.go
@@ -117,6 +117,69 @@
resetExporter(exporter1)
}
+func TestFlushNoOpForIntervalReader(t *testing.T) {
+ ir1 = createAndStart(exporter1, duration1, t)
+
+ gaugeEntry.Set(1)
+
+ // since IR is not stopped, flush does nothing
+ ir1.Flush()
+
+ // expect no data points
+ checkExportedCount(exporter1, 0, t)
+ checkExportedMetricDesc(exporter1, "active_request", t)
+ ir1.Stop()
+ resetExporter(exporter1)
+}
+
+func TestFlushAllowMultipleForIntervalReader(t *testing.T) {
+ ir1 = createAndStart(exporter1, duration1, t)
+
+ gaugeEntry.Set(1)
+
+ ir1.Stop()
+ ir1.Flush()
+
+ // metric is still coming in
+ gaugeEntry.Add(1)
+
+ // one more flush after IR stopped
+ ir1.Flush()
+
+ // expect 2 data point, one from each flush
+ checkExportedCount(exporter1, 2, t)
+ checkExportedValues(exporter1, []int64{1, 2}, t)
+ checkExportedMetricDesc(exporter1, "active_request", t)
+
+ resetExporter(exporter1)
+}
+
+func TestFlushRestartForIntervalReader(t *testing.T) {
+ ir1 = createAndStart(exporter1, duration1, t)
+
+ gaugeEntry.Set(1)
+ ir1.Stop()
+ ir1.Flush()
+
+ // restart the IR
+ err := ir1.Start()
+ if err != nil {
+ t.Fatalf("error starting reader %v\n", err)
+ }
+
+ gaugeEntry.Add(1)
+
+ ir1.Stop()
+ ir1.Flush()
+
+ // expect 2 data point, one from each flush
+ checkExportedCount(exporter1, 2, t)
+ checkExportedValues(exporter1, []int64{1, 2}, t)
+ checkExportedMetricDesc(exporter1, "active_request", t)
+
+ resetExporter(exporter1)
+}
+
func TestProducerWithIntervalReaderStop(t *testing.T) {
ir1 = createAndStart(exporter1, duration1, t)
ir1.Stop()
@@ -166,7 +229,10 @@
func TestIntervalReaderMultipleStart(t *testing.T) {
ir1 = createAndStart(exporter1, duration1, t)
- ir1.Start()
+ err := ir1.Start()
+ if err == nil {
+ t.Fatalf("expected error but got nil\n")
+ }
gaugeEntry.Add(1)
