Fixes MaxCollectDuration log in the streaming proxy

intelsdi-x · Jun 16, 2017 · 0e78d6b · 0e78d6b
1 parent 7d272ce
commit 0e78d6b
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 5 deletions.
diff --git a/examples/snap-plugin-collector-rand-streaming/rand/rand.go b/examples/snap-plugin-collector-rand-streaming/rand/rand.go
@@ -126,7 +126,7 @@ func (r *RandCollector) streamIt(ch chan []plugin.Metric, err chan string) {
 			}
 		}
 		ch <- metrics
-		time.Sleep(time.Second * time.Duration(rand.Int63n(10)))
+		time.Sleep(time.Millisecond * time.Duration(rand.Int63n(1000)))
 	}
 }
 

diff --git a/v1/plugin/stream_collector_proxy.go b/v1/plugin/stream_collector_proxy.go
@@ -101,6 +101,7 @@ func (p *StreamProxy) metricSend(ch chan []Metric, stream rpc.StreamCollector_St
 	).Debug("sending metrics")
 	metrics := []*rpc.Metric{}
 
+	afterCollectDuration := time.After(p.maxCollectDuration)
 	for {
 		select {
 		case mts := <-ch:
@@ -128,11 +129,14 @@ func (p *StreamProxy) metricSend(ch chan []Metric, stream rpc.StreamCollector_St
 			if p.maxMetricsBuffer == 0 {
 				sendReply(metrics, stream)
 				metrics = []*rpc.Metric{}
+				afterCollectDuration = time.After(p.maxCollectDuration)
 			}
-		case <-time.After(p.maxCollectDuration):
+
+		case <-afterCollectDuration:
 			// send metrics if maxCollectDuration is reached
 			sendReply(metrics, stream)
 			metrics = []*rpc.Metric{}
+			afterCollectDuration = time.After(p.maxCollectDuration)
 		case <-stream.Context().Done():
 			return
 		}
@@ -169,7 +173,7 @@ func (p *StreamProxy) streamRecv(ch chan []Metric, stream rpc.StreamCollector_St
 				if s.MaxCollectDuration > 0 {
 					logger.WithFields(log.Fields{
 						"option": "max-collect-duration",
-						"value":  fmt.Sprintf("seconds %v", time.Duration(s.MaxCollectDuration).Seconds()),
+						"value":  fmt.Sprintf("%v seconds", time.Duration(s.MaxCollectDuration).Seconds()),
 					}).Debug("setting max collect duration option")
 					p.setMaxCollectDuration(time.Duration(s.MaxCollectDuration))
 				}

diff --git a/v1/plugin/stream_collector_proxy_test.go b/v1/plugin/stream_collector_proxy_test.go
@@ -170,8 +170,8 @@ func TestStreamMetrics(t *testing.T) {
 			Convey("get buffered metrics through stream proxy when maxMetricsBuffer is reached", func() {
 				Convey("when maxMetricsBuffer is reached", func() {
 					So(pl.outMetric, ShouldNotBeNil)
-					// Send metrics down to channel every 100 ms
-					pl.doAction(time.Millisecond*100, metrics)
+					// Send metrics down to channel every 20 ms
+					pl.doAction(time.Millisecond*20, metrics)
 					select {
 					case mts := <-sendChan:
 						// Success! we got something....
-Original file line number
+Diff line change
@@ Expand Up @@
     			}
     		}
     		ch <- metrics
-    		time.Sleep(time.Second * time.Duration(rand.Int63n(10)))
+    		time.Sleep(time.Millisecond * time.Duration(rand.Int63n(1000)))
     	}
     }
@@ Expand Down @@