tracer: enable stats flushing when Flush() is called

ddtrace/tracer/metrics.go

@@ -23,6 +23,7 @@ type statsdClient interface {
 	Count(name string, value int64, tags []string, rate float64) error
 	Gauge(name string, value float64, tags []string, rate float64) error
 	Timing(name string, value time.Duration, tags []string, rate float64) error
+	Flush() error
 	Close() error
 }
 

ddtrace/tracer/metrics_test.go

@@ -35,6 +35,7 @@ type testStatsdClient struct {
 	waitCh      chan struct{}
 	n           int
 	closed      bool
+	flushed     int
 }
 
 type testStatsdCall struct {
@@ -122,6 +123,13 @@ func (tg *testStatsdClient) addMetric(ct callType, tags []string, c testStatsdCa
 	return nil
 }
 
+func (tg *testStatsdClient) Flush() error {
+	tg.mu.Lock()
+	defer tg.mu.Unlock()
+	tg.flushed++
+	return nil
+}
+
 func (tg *testStatsdClient) Close() error {
 	tg.closed = true
 	return nil

ddtrace/tracer/tracer.go

@@ -311,6 +311,8 @@ func (t *tracer) worker(tick <-chan time.Time) {
 		case done := <-t.flush:
 			t.statsd.Incr("datadog.tracer.flush_triggered", []string{"reason:invoked"}, 1)
 			t.traceWriter.flush()
+			t.statsd.Flush()
+			t.stats.flushAndSend(time.Now(), withoutCurrentBucket)
 			// TODO(x): In reality, the traceWriter.flush() call is not synchronous
 			// when using the agent traceWriter. However, this functionnality is used
 			// in Lambda so for that purpose this mechanism should suffice.

ddtrace/tracer/tracer_test.go

@@ -1932,9 +1932,20 @@ func (w *testTraceWriter) Flushed() []*span {
 
 func TestFlush(t *testing.T) {
 	tr, _, _, stop := startTestTracer(t)
+	defer stop()
+
 	tw := newTestTraceWriter()
 	tr.traceWriter = tw
-	defer stop()
+
+	ts := &testStatsdClient{}
+	tr.statsd = ts
+
+	transport := newDummyTransport()
+	c := newConcentrator(&config{transport: transport}, defaultStatsBucketSize)
+	tr.stats = c
+	c.Start()
+	defer c.Stop()
+
 	tr.StartSpan("op").Finish()
 	timeout := time.After(time.Second)
 loop:
@@ -1950,9 +1961,23 @@ loop:
 			time.Sleep(time.Millisecond)
 		}
 	}
+	as := &aggregableSpan{
+		key: aggregation{
+			Name: "http.request",
+		},
+		// Start must be older than latest bucket to get flushed
+		Start:    time.Now().UnixNano() - 3*defaultStatsBucketSize,
+		Duration: 1,
+	}
+	c.add(as)
+
 	assert.Len(t, tw.Flushed(), 0)
+	assert.Zero(t, ts.flushed)
+	assert.Len(t, transport.Stats(), 0)
 	tr.flushSync()
 	assert.Len(t, tw.Flushed(), 1)
+	assert.Equal(t, ts.flushed, 1)
+	assert.Len(t, transport.Stats(), 1)
 }
 
 func TestTakeStackTrace(t *testing.T) {