// Flush sends metrics to Datadog
170705
func (dd *DatadogMetricSink) Flush(ctx context.Context, interMetrics []samplers.InterMetric) error {
	span, _ := trace.StartSpanFromContext(ctx, "")
	defer span.ClientFinish(dd.traceClient)

	ddmetrics, checks := dd.finalizeMetrics(interMetrics)

	if len(checks) != 0 {
		// this endpoint is not documented to take an array... but it does
		// another curious constraint of this endpoint is that it does not
		// support "Content-Encoding: deflate"
		err := vhttp.PostHelper(context.TODO(), dd.HTTPClient, dd.traceClient, http.MethodPost, fmt.Sprintf("%s/api/v1/check_run?api_key=%s", dd.DDHostname, dd.APIKey), checks, "flush_checks", false, map[string]string{"sink": "datadog"}, dd.log)
		if err == nil {
			dd.log.WithField("checks", len(checks)).Info("Completed flushing service checks to Datadog")
		} else {
			dd.log.WithFields(logrus.Fields{
				"checks":        len(checks),
				logrus.ErrorKey: err}).Warn("Error flushing checks to Datadog")
		}
	}

	// break the metrics into chunks of approximately equal size, such that
	// each chunk is less than the limit
	// we compute the chunks using rounding-up integer division
	workers := ((len(ddmetrics) - 1) / dd.flushMaxPerBody) + 1
	chunkSize := ((len(ddmetrics) - 1) / workers) + 1
	dd.log.WithField("workers", workers).Debug("Worker count chosen")
	dd.log.WithField("chunkSize", chunkSize).Debug("Chunk size chosen")
	var wg sync.WaitGroup
	flushStart := time.Now()
	for i := 0; i < workers; i++ {
		chunk := ddmetrics[i*chunkSize:]
		if i < workers-1 {
			// trim to chunk size unless this is the last one
			chunk = chunk[:chunkSize]
		}
		wg.Add(1)
		go dd.flushPart(span.Attach(ctx), chunk, &wg)
	}
	wg.Wait()
	tags := map[string]string{"sink": dd.Name()}
	span.Add(
		ssf.Timing(sinks.MetricKeyMetricFlushDuration, time.Since(flushStart), time.Nanosecond, tags),
		ssf.Count(sinks.MetricKeyTotalMetricsFlushed, float32(len(ddmetrics)), tags),
	)
	dd.log.WithField("metrics", len(ddmetrics)).Info("Completed flush to Datadog")
	return nil
}
233209
func (b *NoStorage) Flush() error {
	if f, ok := b.w.(http.Flusher); ok {
		f.Flush()
	}
	return nil
}
1281991
func (b *BulkIndexer) Flush() {
	b.mu.Lock()
	if b.docCt > 0 {
		b.send(b.buf)
	}
	b.mu.Unlock()
}
980812
func (w *TimeWait) Flush() {
	w.doonce.Do(func() {
		close(w.closer)
		w.action.Fullfill(0)
		atomic.StoreInt64(&w.hits, 0)
	})
}
1012976
func (c *Client) Flush() {
	c.initBundler()
	c.bundler.Flush()
	c.outstanding.Wait()
}
1143762
func (bg *Writer) Flush() error {
	if bg.closed {
		return ErrClosed
	}
	if err := bg.Error(); err != nil {
		return err
	}

	if bg.active.next == 0 {
		return nil
	}

	var c *compressor
	c, bg.active = bg.active, <-bg.waiting
	bg.queue <- c
	bg.qwg.Add(1)
	go c.writeBlock()

	return bg.Error()
}
1039182
func (w *gzipResponseWriter) Flush() {
	if f, ok := w.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	} else {
		panic(httpserver.NonFlusherError{Underlying: w.ResponseWriter}) // should be recovered at the beginning of middleware stack
	}
}
7052
func (w *Writer) Flush() error {
	return w.cw.w.(*bufio.Writer).Flush()
}
1247189
func (sl *DiskSessionLogger) flush() error {
	var err, err2 error

	if sl.RecordSessions && sl.chunksFile != nil {
		err = sl.chunksFile.Flush()
	}
	if sl.eventsFile != nil {
		err2 = sl.eventsFile.Flush()
	}
	return trace.NewAggregate(err, err2)
}
279391
func (r *responseWriter) Flush() {
	if f, ok := r.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}
1310134
func (b *Buffer) Flush() (size int, err error) {
	for _, buf := range b.buffer {
		if size, err = b.write(buf); err != nil {
			return
		}
	}

	b.buffer = b.buffer[:0]
	return
}
1284994
func (i *Handle) Flush() error {
	_, err := i.doCmdWithoutAttr(ipvsCmdFlush)
	return err
}
14133
func (st *CookieSessionStore) Flush() {
	st.lock.Lock()
	defer st.lock.Unlock()
	st.values = make(map[interface{}]interface{})
}
793941
func (logger *Logger) Flush() {
	b := &backgroundFlusher{
		logger:     logger,
		updateChan: make(chan struct{}),
	}
	transitChannel <- b
	<-b.updateChan
}
1331256
func (w *PktlineWriter) Flush() error {
	if w == nil {
		return nil
	}

	if _, err := w.flush(); err != nil {
		return err
	}

	if err := w.pl.writeFlush(); err != nil {
		return err
	}

	return nil
}
1300880
func (s *KVStore) Flush(ctx context.Context) {
	_ = s.db.Update(
		func(tx *bolt.Tx) error {
			return tx.ForEach(func(name []byte, b *bolt.Bucket) error {
				s.cleanBucket(tx, b)
				return nil
			})
		},
	)
}
1221567
func (gs *GRPCSpanSink) Flush() {
	samples := &ssf.Samples{}
	samples.Add(
		ssf.Count(
			sinks.MetricKeyTotalSpansFlushed,
			float32(atomic.SwapUint32(&gs.sentCount, 0)),
			map[string]string{"sink": gs.Name()}),
		ssf.Count(
			sinks.MetricKeyTotalSpansDropped,
			float32(atomic.SwapUint32(&gs.dropCount, 0)),
			map[string]string{"sink": gs.Name()},
		),
	)

	metrics.Report(gs.traceClient, samples)
	return
}