// FlushOtherSamples serializes Events or Service Checks directly to datadog.
// May make 2 external calls to the datadog client.
480626
func (dd *DatadogMetricSink) FlushOtherSamples(ctx context.Context, samples []ssf.SSFSample) {

	events := []DDEvent{}

	span, _ := trace.StartSpanFromContext(ctx, "")
	defer span.ClientFinish(dd.traceClient)

	// fill in the default hostname for packets that didn't set it
	for _, sample := range samples {

		if _, ok := sample.Tags[dogstatsd.EventIdentifierKey]; ok {
			// This is an event!
			ret := DDEvent{
				Title:     sample.Name,
				Text:      sample.Message,
				Timestamp: sample.Timestamp,
				Priority:  "normal",
				AlertType: "info",
			}

			// Defensively copy the tags that came in
			tags := map[string]string{}
			for k, v := range sample.Tags {
				tags[k] = v
			}
			// Remove the tag that flagged this as an event
			delete(tags, dogstatsd.EventIdentifierKey)

			// The parser uses special tags to encode the fields for us from DogStatsD
			// that don't fit into a normal SSF Sample. We'll hunt for each one and
			// delete the tag if we find it.
			if v, ok := tags[dogstatsd.EventAggregationKeyTagKey]; ok {
				ret.Aggregation = v
				delete(tags, dogstatsd.EventAggregationKeyTagKey)
			}
			if v, ok := tags[dogstatsd.EventPriorityTagKey]; ok {
				ret.Priority = v
				delete(tags, dogstatsd.EventPriorityTagKey)
			}
			if v, ok := tags[dogstatsd.EventSourceTypeTagKey]; ok {
				ret.Source = v
				delete(tags, dogstatsd.EventSourceTypeTagKey)
			}
			if v, ok := tags[dogstatsd.EventAlertTypeTagKey]; ok {
				ret.AlertType = v
				delete(tags, dogstatsd.EventAlertTypeTagKey)
			}
			if v, ok := tags[dogstatsd.EventHostnameTagKey]; ok {
				ret.Hostname = v
				delete(tags, dogstatsd.EventHostnameTagKey)
			} else {
				// Default hostname since there isn't one
				ret.Hostname = dd.hostname
			}
			// Do our last bit of tag housekeeping
			finalTags := []string{}
			for k, v := range tags {
				finalTags = append(finalTags, fmt.Sprintf("%s:%s", k, v))
			}

			ret.Tags = append(finalTags, dd.tags...)
			events = append(events, ret)
		} else {
			dd.log.Warn("Received an SSF Sample that wasn't an event or service check, ack!")
		}
	}

	if len(events) != 0 {
		// this endpoint is not documented at all, its existence is only known from
		// the official dd-agent
		// we don't actually pass all the body keys that dd-agent passes here... but
		// it still works
		err := vhttp.PostHelper(context.Background(), dd.HTTPClient, dd.traceClient, http.MethodPost, fmt.Sprintf("%s/intake?api_key=%s", dd.DDHostname, dd.APIKey), map[string]map[string][]DDEvent{
			"events": {
				"api": events,
			},
		}, "flush_events", true, map[string]string{"sink": "datadog"}, dd.log)

		if err == nil {
			dd.log.WithField("events", len(events)).Info("Completed flushing events to Datadog")
		} else {
			dd.log.WithFields(logrus.Fields{
				"events":        len(events),
				logrus.ErrorKey: err}).Warn("Error flushing events to Datadog")
		}
	}
}
1155695
func (ew *EventWorker) Flush() []ssf.SSFSample {
	ew.mutex.Lock()

	retsamples := ew.samples
	// these slices will be allocated again at append time
	ew.samples = nil

	ew.mutex.Unlock()
	if len(retsamples) != 0 {
		ew.stats.Count("worker.other_samples_flushed_total", int64(len(retsamples)), nil, 1.0)
	}
	return retsamples
}
1086892
func (s *rateLimitingSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
	return s.rateLimiter.CheckCredit(1.0), s.tags
}
1233978
func (s *GetTraceSummariesInput) SetSamplingStrategy(v *SamplingStrategy) *GetTraceSummariesInput {
	s.SamplingStrategy = v
	return s
}
202506
func (r *InMemorySpanRecorder) GetSampledSpans() []RawSpan {
	r.RLock()
	defer r.RUnlock()
	spans := make([]RawSpan, 0, len(r.spans))
	for _, span := range r.spans {
		if span.Context.Sampled {
			spans = append(spans, span)
		}
	}
	return spans
}
364109
func (s *FloatEncoder) Flush() {
	if !s.finished {
		// write an end-of-stream record
		s.finished = true
		s.Write(math.NaN())
		s.bw.Flush(bitstream.Zero)
	}
}
957770
func (c *syncClient) Flushall() (err Error) {
	_, err = c.conn.ServiceRequest(&FLUSHALL, [][]byte{})
	return
}
1259948
func (fw *Firewaller) flushUnits(unitds []*unitData) error {
	machineds := map[names.MachineTag]*machineData{}
	for _, unitd := range unitds {
		machineds[unitd.machined.tag] = unitd.machined
	}
	for _, machined := range machineds {
		if err := fw.flushMachine(machined); err != nil {
			return err
		}
	}
	return nil
}
1103965
func (w *SegmentWAL) LogSamples(samples []RefSample) error {
	buf := w.getBuffer()

	flag := w.encodeSamples(buf, samples)

	w.mtx.Lock()
	defer w.mtx.Unlock()

	err := w.write(WALEntrySamples, flag, buf.Get())

	w.putBuffer(buf)

	if err != nil {
		return errors.Wrap(err, "log series")
	}
	tf := w.head()

	for _, s := range samples {
		if tf.maxTime < s.T {
			tf.maxTime = s.T
		}
	}
	return nil
}
1352363
func (self *FSock) filterEvents(filters map[string][]string) error {
	if len(filters) == 0 {
		return nil
	}
	filters["Event-Name"] = append(filters["Event-Name"], "BACKGROUND_JOB") // for bgapi
	for hdr, vals := range filters {
		for _, val := range vals {
			cmd := "filter " + hdr + " " + val + "\n\n"
			self.send(cmd)
			if rply, err := self.readHeaders(); err != nil {
				return err
			} else if !strings.Contains(rply, "Reply-Text: +OK") {
				return fmt.Errorf("Unexpected filter-events reply received: <%s>", rply)
			}
		}
	}
	return nil
}
545991
func (s *GetSamplingStatisticSummariesOutput) SetSamplingStatisticSummaries(v []*SamplingStatisticSummary) *GetSamplingStatisticSummariesOutput {
	s.SamplingStatisticSummaries = v
	return s
}
573215
func (w *Wav) ReadSamples(n int) (interface{}, error) {
	var data interface{}
	switch w.AudioFormat {
	case wavFormatPCM:
		switch w.BitsPerSample {
		case 8:
			data = make([]uint8, n)
		case 16:
			data = make([]int16, n)
		default:
			return nil, fmt.Errorf("wav: unknown bits per sample: %v", w.BitsPerSample)
		}
	case wavFormatIEEEFloat:
		data = make([]float32, n)
	default:
		return nil, fmt.Errorf("wav: unknown audio format")
	}
	if err := binary.Read(w.r, binary.LittleEndian, data); err != nil {
		return nil, err
	}
	return data, nil
}
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
}
1321246
func (hook *SentryHook) SetSampleRate(rate float32) error {
	return hook.client.SetSampleRate(rate)
}
436565
func (s *GetSamplingRulesOutput) SetSamplingRuleRecords(v []*SamplingRuleRecord) *GetSamplingRulesOutput {
	s.SamplingRuleRecords = v
	return s
}
1057081
func (p *PerKeyThroughput) GetSampleRate(key string) int {
	p.lock.Lock()
	defer p.lock.Unlock()
	p.currentCounts[key]++
	if rate, found := p.savedSampleRates[key]; found {
		return rate
	}
	return 1
}
55931
func (c *HeapProfiler) StartSamplingWithParams(v *HeapProfilerStartSamplingParams) (*gcdmessage.ChromeResponse, error) {
	return gcdmessage.SendDefaultRequest(c.target, c.target.GetSendCh(), &gcdmessage.ParamRequest{Id: c.target.GetId(), Method: "HeapProfiler.startSampling", Params: v})
}