func (a *Agent) Process(p *api.Payload) {
	if len(p.Traces) == 0 {
		log.Debugf("Skipping received empty payload")
		return
	}
	defer timing.Since("datadog.trace_agent.internal.process_payload_ms", time.Now())
	ts := p.Source
	ss := new(writer.SampledSpans)
	var envtraces []stats.EnvTrace
	a.PrioritySampler.CountClientDroppedP0s(p.ClientDroppedP0s)
	for _, t := range p.Traces {
		if len(t) == 0 {
			log.Debugf("Skipping received empty trace")
			continue
		}

		tracen := int64(len(t))
		atomic.AddInt64(&ts.SpansReceived, tracen)
		err := normalizeTrace(p.Source, t)
		if err != nil {
			log.Debug("Dropping invalid trace: %s", err)
			atomic.AddInt64(&ts.SpansDropped, tracen)
			continue
		}

		// Root span is used to carry some trace-level metadata, such as sampling rate and priority.
		root := traceutil.GetRoot(t)

		if !a.Blacklister.Allows(root) {
			log.Debugf("Trace rejected by blacklister. root: %v", root)
			atomic.AddInt64(&ts.TracesFiltered, 1)
			atomic.AddInt64(&ts.SpansFiltered, tracen)
			continue
		}

		if filteredByTags(root, a.conf.RequireTags, a.conf.RejectTags) {
			log.Debugf("Trace rejected as it fails to meet tag requirements. root: %v", root)
			atomic.AddInt64(&ts.TracesFiltered, 1)
			atomic.AddInt64(&ts.SpansFiltered, tracen)
			continue
		}

		// Extra sanitization steps of the trace.
		for _, span := range t {
			for k, v := range a.conf.GlobalTags {
				traceutil.SetMeta(span, k, v)
			}
			a.obfuscator.Obfuscate(span)
			Truncate(span)
			if p.ClientComputedTopLevel {
				traceutil.UpdateTracerTopLevel(span)
			}
		}
		a.Replacer.Replace(t)

		{
			// this section sets up any necessary tags on the root:
			clientSampleRate := sampler.GetGlobalRate(root)
			sampler.SetClientRate(root, clientSampleRate)

			if ratelimiter := a.Receiver.RateLimiter; ratelimiter.Active() {
				rate := ratelimiter.RealRate()
				sampler.SetPreSampleRate(root, rate)
			}
			if p.ContainerTags != "" {
				traceutil.SetMeta(root, tagContainersTags, p.ContainerTags)
			}
		}
		if !p.ClientComputedTopLevel {
			// Figure out the top-level spans now as it involves modifying the Metrics map
			// which is not thread-safe while samplers and Concentrator might modify it too.
			traceutil.ComputeTopLevel(t)
		}

		env := a.conf.DefaultEnv
		if v := traceutil.GetEnv(t); v != "" {
			// this trace has a user defined env.
			env = v
		}
		pt := ProcessedTrace{
			Trace:            t,
			WeightedTrace:    stats.NewWeightedTrace(t, root),
			Root:             root,
			Env:              env,
			ClientDroppedP0s: p.ClientDroppedP0s > 0,
		}

		events, keep := a.sample(ts, pt)
		if !p.ClientComputedStats {
			if envtraces == nil {
				envtraces = make([]stats.EnvTrace, 0, len(p.Traces))
			}
			envtraces = append(envtraces, stats.EnvTrace{
				Trace: pt.WeightedTrace,
				Env:   pt.Env,
			})
		}
		// TODO(piochelepiotr): Maybe we can skip some computation if stats are computed in the tracer and the trace is droped.
		if keep {
			ss.Traces = append(ss.Traces, traceutil.APITrace(t))
			ss.Size += t.Msgsize()
			ss.SpanCount += int64(len(t))
		}
		if len(events) > 0 {
			ss.Events = append(ss.Events, events...)
			ss.Size += pb.Trace(events).Msgsize()
		}
		if ss.Size > writer.MaxPayloadSize {
			a.TraceWriter.In <- ss
			ss = new(writer.SampledSpans)
		}
	}
	if ss.Size > 0 {
		a.TraceWriter.In <- ss
	}
	if len(envtraces) > 0 {
		in := stats.Input{Traces: envtraces}
		if !features.Has("disable_cid_stats") && a.conf.IsFargate {
			// only allow the ContainerID stats dimension if we're in a Fargate instance
			// and it's not prohibited by the disable_cid_stats feature flag.
			in.ContainerID = p.ContainerID
		}
		a.Concentrator.In <- in
	}
}
// Process is the default work unit that receives a trace, transforms it and
// passes it downstream.
Process is the default work unit that receives a trace, transforms it and
passes it downstream.
// Process is the default work unit that receives a trace transforms it and // passes it downstream .
func (a *Agent) Process(p *api.Payload) {
	if len(p.Traces) == 0 {
		log.Debugf("Skipping received empty payload")
		return
	}
	defer timing.Since("datadog.trace_agent.internal.process_payload_ms", time.Now())
	ts := p.Source
	ss := new(writer.SampledSpans)
	var envtraces []stats.EnvTrace
	a.PrioritySampler.CountClientDroppedP0s(p.ClientDroppedP0s)
	for _, t := range p.Traces {
		if len(t) == 0 {
			log.Debugf("Skipping received empty trace")
			continue
		}
		tracen := int64(len(t))
		atomic.AddInt64(&ts.SpansReceived, tracen)
		err := normalizeTrace(p.Source, t)
		if err != nil {
			log.Debug("Dropping invalid trace: %s", err)
			atomic.AddInt64(&ts.SpansDropped, tracen)
			continue
		}
		root := traceutil.GetRoot(t)
		if !a.Blacklister.Allows(root) {
			log.Debugf("Trace rejected by blacklister. root: %v", root)
			atomic.AddInt64(&ts.TracesFiltered, 1)
			atomic.AddInt64(&ts.SpansFiltered, tracen)
			continue
		}
		if filteredByTags(root, a.conf.RequireTags, a.conf.RejectTags) {
			log.Debugf("Trace rejected as it fails to meet tag requirements. root: %v", root)
			atomic.AddInt64(&ts.TracesFiltered, 1)
			atomic.AddInt64(&ts.SpansFiltered, tracen)
			continue
		}
		for _, span := range t {
			for k, v := range a.conf.GlobalTags {
				traceutil.SetMeta(span, k, v)
			}
			a.obfuscator.Obfuscate(span)
			Truncate(span)
			if p.ClientComputedTopLevel {
				traceutil.UpdateTracerTopLevel(span)
			}
		}
		a.Replacer.Replace(t)
		{
			clientSampleRate := sampler.GetGlobalRate(root)
			sampler.SetClientRate(root, clientSampleRate)
			if ratelimiter := a.Receiver.RateLimiter; ratelimiter.Active() {
				rate := ratelimiter.RealRate()
				sampler.SetPreSampleRate(root, rate)
			}
			if p.ContainerTags != "" {
				traceutil.SetMeta(root, tagContainersTags, p.ContainerTags)
			}
		}
		if !p.ClientComputedTopLevel {
			traceutil.ComputeTopLevel(t)
		}
		env := a.conf.DefaultEnv
		if v := traceutil.GetEnv(t); v != "" {
			env = v
		}
		pt := ProcessedTrace{
			Trace:            t,
			WeightedTrace:    stats.NewWeightedTrace(t, root),
			Root:             root,
			Env:              env,
			ClientDroppedP0s: p.ClientDroppedP0s > 0,
		}
		events, keep := a.sample(ts, pt)
		if !p.ClientComputedStats {
			if envtraces == nil {
				envtraces = make([]stats.EnvTrace, 0, len(p.Traces))
			}
			envtraces = append(envtraces, stats.EnvTrace{
				Trace: pt.WeightedTrace,
				Env:   pt.Env,
			})
		}
		if keep {
			ss.Traces = append(ss.Traces, traceutil.APITrace(t))
			ss.Size += t.Msgsize()
			ss.SpanCount += int64(len(t))
		}
		if len(events) > 0 {
			ss.Events = append(ss.Events, events...)
			ss.Size += pb.Trace(events).Msgsize()
		}
		if ss.Size > writer.MaxPayloadSize {
			a.TraceWriter.In <- ss
			ss = new(writer.SampledSpans)
		}
	}
	if ss.Size > 0 {
		a.TraceWriter.In <- ss
	}
	if len(envtraces) > 0 {
		in := stats.Input{Traces: envtraces}
		if !features.Has("disable_cid_stats") && a.conf.IsFargate {
			in.ContainerID = p.ContainerID
		}
		a.Concentrator.In <- in
	}
}
func ( a * Agent ) Process ( p * api . Payload ) { if len ( p . Traces ) == 0 { log . Debugf ( " " ) return } defer timing . Since ( " " , time . Now ( ) ) ts := p . Source ss := new ( writer . SampledSpans ) var envtraces [ ] stats . EnvTrace a . PrioritySampler . CountClientDroppedP0s ( p . ClientDroppedP0s ) for _ , t := range p . Traces { if len ( t ) == 0 { log . Debugf ( " " ) continue } tracen := int64 ( len ( t ) ) atomic . AddInt64 ( & ts . SpansReceived , tracen ) err := normalizeTrace ( p . Source , t ) if err != nil { log . Debug ( " " , err ) atomic . AddInt64 ( & ts . SpansDropped , tracen ) continue } root := traceutil . GetRoot ( t ) if ! a . Blacklister . Allows ( root ) { log . Debugf ( " " , root ) atomic . AddInt64 ( & ts . TracesFiltered , 1 ) atomic . AddInt64 ( & ts . SpansFiltered , tracen ) continue } if filteredByTags ( root , a . conf . RequireTags , a . conf . RejectTags ) { log . Debugf ( " " , root ) atomic . AddInt64 ( & ts . TracesFiltered , 1 ) atomic . AddInt64 ( & ts . SpansFiltered , tracen ) continue } for _ , span := range t { for k , v := range a . conf . GlobalTags { traceutil . SetMeta ( span , k , v ) } a . obfuscator . Obfuscate ( span ) Truncate ( span ) if p . ClientComputedTopLevel { traceutil . UpdateTracerTopLevel ( span ) } } a . Replacer . Replace ( t ) { clientSampleRate := sampler . GetGlobalRate ( root ) sampler . SetClientRate ( root , clientSampleRate ) if ratelimiter := a . Receiver . RateLimiter ; ratelimiter . Active ( ) { rate := ratelimiter . RealRate ( ) sampler . SetPreSampleRate ( root , rate ) } if p . ContainerTags != " " { traceutil . SetMeta ( root , tagContainersTags , p . ContainerTags ) } } if ! p . ClientComputedTopLevel { traceutil . ComputeTopLevel ( t ) } env := a . conf . DefaultEnv if v := traceutil . GetEnv ( t ) ; v != " " { env = v } pt := ProcessedTrace { Trace : t , WeightedTrace : stats . NewWeightedTrace ( t , root ) , Root : root , Env : env , ClientDroppedP0s : p . ClientDroppedP0s > 0 , } events , keep := a . sample ( ts , pt ) if ! p . ClientComputedStats { if envtraces == nil { envtraces = make ( [ ] stats . EnvTrace , 0 , len ( p . Traces ) ) } envtraces = append ( envtraces , stats . EnvTrace { Trace : pt . WeightedTrace , Env : pt . Env , } ) } if keep { ss . Traces = append ( ss . Traces , traceutil . APITrace ( t ) ) ss . Size += t . Msgsize ( ) ss . SpanCount += int64 ( len ( t ) ) } if len ( events ) > 0 { ss . Events = append ( ss . Events , events ... ) ss . Size += pb . Trace ( events ) . Msgsize ( ) } if ss . Size > writer . MaxPayloadSize { a . TraceWriter . In <- ss ss = new ( writer . SampledSpans ) } } if ss . Size > 0 { a . TraceWriter . In <- ss } if len ( envtraces ) > 0 { in := stats . Input { Traces : envtraces } if ! features . Has ( " " ) && a . conf . IsFargate { in . ContainerID = p . ContainerID } a . Concentrator . In <- in } }
Process is the default work unit that receives a trace, transforms it and
passes it downstream.
Process is the default work unit that receives a trace transforms it and passes it downstream .
// Root span is used to carry some trace-level metadata, such as sampling rate and priority. // Extra sanitization steps of the trace. // this section sets up any necessary tags on the root: // Figure out the top-level spans now as it involves modifying the Metrics map // which is not thread-safe while samplers and Concentrator might modify it too. // this trace has a user defined env. // TODO(piochelepiotr): Maybe we can skip some computation if stats are computed in the tracer and the trace is droped. // only allow the ContainerID stats dimension if we're in a Fargate instance // and it's not prohibited by the disable_cid_stats feature flag.