trace2emitotlp.go

package trace2receiver

import (
	"encoding/json"
	"fmt"
	"runtime"

	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/ptrace"
	"go.opentelemetry.io/otel"
	semconv "go.opentelemetry.io/otel/semconv/v1.14.0"
)

func (tr2 *trace2Dataset) insertResourceServiceFields(resourceAttrs pcommon.Map) {
	// The SemConv `service.namespace`, `service.name`, `service.version`,
	// and `service.instance.id` fields are somewhat ill-defined in our
	// case.  They describe the application generating the telemetry and
	// an organizational/ownership group that multiple services operate
	// under.
	//
	// This `trace2receiver` component is just a relay/proxy for the
	// actual telemetry being generated by commands like `git.exe` or
	// `gcm.exe`.  That is, this component is not generating original
	// telemetry about itself.  So we adapt the definitions of these
	// fields a bit here.

	// [1] Claim a namespace for all of the things that we will proxy
	// since the Git commands don't know anything about any of the OTEL
	// and OTLP stuff.
	//
	// TODO Let's make the namespace a constant for now.  Later, we can
	// TODO consider if we should read it from the `config.yaml` to
	// TODO allow us to fit into our deployment's SemConv scheme.

	resourceAttrs.PutStr(string(semconv.ServiceNamespaceKey), Trace2ServiceNamespace)

	// [2] Use the name of the Git command to define the service name.
	// The OTEL guidelines suggest that this is just the name of the
	// application (a service that may be running on more than one host
	// instances).  However, since some visualization tools automatically
	// group by the service name, just putting `git` or `gcm` in this
	// field doesn't do much for us.  Using the `<name>:<verb>%<mode>`
	// string is better here.  This deviation feels right since an
	// actual service application may define multiple end-points and do
	// many different things.
	//
	// Using the `<name>:<verb>%<mode` as the service name also has the
	// nice property the service name is attached to every region span
	// and that can help in some queries.

	resourceAttrs.PutStr(string(semconv.ServiceNameKey), tr2.process.qualifiedNames.exeVerbMode)

	// [3] Use the Git version number for `service.version` (and not the
	// version number of this component).

	resourceAttrs.PutStr(string(semconv.ServiceVersionKey), tr2.process.exeVersion)

	// [4] The `service.instance.id` field is defined as a way to
	// identify multiple instances of a horizontally-scaled service
	// application.  That doesn't fit our model either, so we bind
	// it to a single invocation of Git and reuse the Trace2 SID.
	// (This is the complete SID with slashes.)

	resourceAttrs.PutStr(string(semconv.ServiceInstanceIDKey), tr2.trace2SID)
}

func (tr2 *trace2Dataset) insertResourceTelemetrySDKFields(resourceAttrs pcommon.Map) {
	// The `telemetry.sdk.*` attributes are also somewhat ill-defined in our
	// case.  They define the telemetry SDK used to capture data recorded by
	// the instrumentation libraries.  The original data was generated in Trace2
	// format by the original process (and Git.exe and GCM.exe have completely
	// different implementations of the Trace2 spec and know nothing about OTEL).
	// So I'm going to interpret the `telemetry.sdk.*` fields as describing the
	// name and version of the OTEL SDK that we are using in this receiver
	// component, since we are the one generating the actual OTLP.

	resourceAttrs.PutStr(string(semconv.TelemetrySDKNameKey), "opentelemetry")
	resourceAttrs.PutStr(string(semconv.TelemetrySDKLanguageKey), "go")
	resourceAttrs.PutStr(string(semconv.TelemetrySDKVersionKey), otel.Version())
}

func (tr2 *trace2Dataset) insertResourceInstrumentationScope(instScope pcommon.InstrumentationScope) {
	// The "instrumentation scope" deals with the instrumentation library
	// that generated the traces.  Again, this is a little ill-defined in
	// our case.  I'm going to record the name and version of this receiver
	// component becase we are relaying/proxying the Trace2 data into OTLP.

	instScope.SetName(Trace2InstrumentationName)
	instScope.SetVersion(Trace2ReceiverVersion)
}

func (tr2 *trace2Dataset) ToTraces(dl FilterDetailLevel) ptrace.Traces {
	pt := ptrace.NewTraces()

	resourceSpans := pt.ResourceSpans().AppendEmpty()
	resourceAttrs := resourceSpans.Resource().Attributes()
	scopes := resourceSpans.ScopeSpans().AppendEmpty()

	tr2.insertResourceServiceFields(resourceAttrs)
	tr2.insertResourceTelemetrySDKFields(resourceAttrs)
	tr2.insertResourceInstrumentationScope(scopes.Scope())

	// For convienence and consistency across various visualization tools,
	// also put some of the above values into our Trace2 attribute bag.

	resourceAttrs.PutStr(string(Trace2CmdVersion), tr2.process.exeVersion)
	resourceAttrs.PutStr(string(Trace2CmdSid), tr2.trace2SID)

	// Create an OTEL span for the entire process (aka the main thread).
	exeSpan := scopes.Spans().AppendEmpty()
	emitProcessSpan(&exeSpan, tr2, dl)

	if WantRegionAndThreadSpans(dl) {
		// Create an OTEL span for the lifetime of each non-main thread.
		for _, th := range tr2.threads {
			thSpan := scopes.Spans().AppendEmpty()
			emitNonMainThreadSpan(&thSpan, th, tr2)
		}

		// Create OTEL spans for all completed regions (from all threads).
		for _, r := range tr2.completedRegions {
			rSpan := scopes.Spans().AppendEmpty()
			emitRegionSpan(&rSpan, r, tr2)
		}
	}

	if WantChildSpans(dl) {
		// Create an OTEL span for each child process that this process created.
		for _, child := range tr2.children {
			childSpan := scopes.Spans().AppendEmpty()
			emitChildSpan(&childSpan, child, tr2)
		}

		for _, exec := range tr2.exec {
			execSpan := scopes.Spans().AppendEmpty()
			emitExecSpan(&execSpan, exec, tr2)
		}
	}

	return pt
}

// The `ptrace.SpanKind` turns out to be an important field
// for some visualization tools and can change how/where data
// is stored in the database.  Or, rather, how some exporters
// export the data.  (This is very much by trial and error.)
//
// [1] Treat all process spans as a `SpanKindServer`.  This
// agrees with our decision to expose the `name:verb%mode` as
// the `service.name`.  Conceptually, these Git commands are
// service points (well, kinda).
//
// [2] Treat all other spans (child, thread, region) as
// `SpanKindInternal` since they are within the process.
// AzureMonitor will label them as `InProc`, for example.
//
// TODO Git commands tend to be synchronous and the caller
// TODO waits for the command to complete, so `SpanKindServer`
// TODO is a better choice than `SpanKindConsumer`.  However,
// TODO we may want to mark commands pushed into the background
// TODO (like FSMonitor) as async consumers.
//
// [3] I considered using `SpanKindClient` for synchronous child
// spans (bounding the `child_start` and `child_exit` events),
// and `SpanKindProducer` for async `child_start` and `child_ready`
// events, but that just added noise to the AzureMonitor graphs.
//
// [4] In AzureMonitor, all internal, producer, and consumer
// spans are stored in the `dependencies` table in the portal view.
// Server spans are stored in the `requests` table.  So you need to
// use `union requests, dependencies | where ...` queries to see
// all of the data.  However, this does let us easily see only
// commands by just selecting on the `requests` table,

// Populate the span with the basic essential values
// required by OTEL.  This includes the OTLP TraceID,
// SpanIDs, and timestamps.
func emitSpanEssentials(span *ptrace.Span, r *TrSpanEssentials, tr2 *trace2Dataset) {

	span.SetName(r.displayName)
	span.SetStartTimestamp(pcommon.NewTimestampFromTime(r.startTime))
	span.SetEndTimestamp(pcommon.NewTimestampFromTime(r.endTime))
	span.SetKind(ptrace.SpanKindInternal)

	span.SetSpanID(r.selfSpanID)
	span.SetParentSpanID(r.parentSpanID)

	span.SetTraceID(tr2.otelTraceID)
}

func emitProcessSpan(span *ptrace.Span, tr2 *trace2Dataset, dl FilterDetailLevel) {
	emitSpanEssentials(span, &tr2.process.mainThread.lifetime, tr2)
	span.SetKind(ptrace.SpanKindServer)

	// TODO Should we set "SpanStatus" based upon the exit code of the process?
	// Possible values are "UNSET", "OK", and "ERROR".
	// This may or may not be the same as the "otel.status_code" tag.
	// The "ptrace" APIs don't currently seem to have a method to do this.
	// The default is "UNSET".

	sm := span.Attributes()
	sm.PutStr(string(Trace2SpanType), "process")

	sm.PutStr(string(Trace2GoArch), runtime.GOARCH)
	sm.PutStr(string(Trace2GoOS), runtime.GOOS)

	for k, v := range tr2.pii {
		sm.PutStr(k, v)
	}

	sm.PutStr(string(Trace2CmdName), tr2.process.qualifiedNames.exe)
	sm.PutStr(string(Trace2CmdNameVerb), tr2.process.qualifiedNames.exeVerb)
	sm.PutStr(string(Trace2CmdNameVerbMode), tr2.process.qualifiedNames.exeVerbMode)
	sm.PutStr(string(Trace2CmdHierarchy), tr2.process.cmdHierarchy)
	sm.PutStr(string(Trace2CmdExitCode), fmt.Sprintf("%d", tr2.process.exeExitCode))

	if len(tr2.process.cmdArgv) > 0 {
		jargs, _ := json.Marshal(tr2.process.cmdArgv)
		sm.PutStr(string(Trace2CmdArgv), string(jargs))
	}

	if WantProcessAncestry(dl) {
		if len(tr2.process.cmdAncestry) > 0 {
			jargs, _ := json.Marshal(tr2.process.cmdAncestry)
			sm.PutStr(string(Trace2CmdAncestry), string(jargs))
		}
	}

	if WantProcessAliases(dl) {
		if len(tr2.process.cmdAliasKey) > 0 {
			sm.PutStr(string(Trace2CmdAliasKey), tr2.process.cmdAliasKey)

			if len(tr2.process.cmdAliasValue) > 0 {
				jargs, _ := json.Marshal(tr2.process.cmdAliasValue)
				sm.PutStr(string(Trace2CmdAliasValue), string(jargs))
			}
		}
	}

	if len(tr2.process.exeErrorFmt) > 0 {
		sm.PutStr(string(Trace2CmdErrFmt), tr2.process.exeErrorFmt)
	}
	if len(tr2.process.exeErrorMsg) > 0 {
		sm.PutStr(string(Trace2CmdErrMsg), tr2.process.exeErrorMsg)
	}

	if tr2.process.repoSet != nil && len(tr2.process.repoSet) > 0 {
		jargs, _ := json.Marshal(tr2.process.repoSet)
		sm.PutStr(string(Trace2RepoSet), string(jargs))
	}

	if tr2.process.paramSetValues != nil && len(tr2.process.paramSetValues) > 0 {
		jargs, _ := json.Marshal(tr2.process.paramSetValues)
		sm.PutStr(string(Trace2ParamSet), string(jargs))
	}

	if WantMainThreadTimersAndCounters(dl) {
		// Emit per-thread counters and timers for the main thread because
		// it is not handled by `emitNonMainThreadSpan()`.
		if tr2.process.mainThread.timers != nil {
			jargs, _ := json.Marshal(tr2.process.mainThread.timers)
			sm.PutStr(string(Trace2ThreadTimers), string(jargs))
		}
		if tr2.process.mainThread.counters != nil {
			jargs, _ := json.Marshal(tr2.process.mainThread.counters)
			sm.PutStr(string(Trace2ThreadCounters), string(jargs))
		}
	}

	if WantProcessTimersCountersAndData(dl) {
		if tr2.process.dataValues != nil && len(tr2.process.dataValues) > 0 {
			jargs, _ := json.Marshal(tr2.process.dataValues)
			sm.PutStr(string(Trace2ProcessData), string(jargs))
		}
		if tr2.process.timers != nil {
			jargs, _ := json.Marshal(tr2.process.timers)
			sm.PutStr(string(Trace2ProcessTimers), string(jargs))
		}
		if tr2.process.counters != nil {
			jargs, _ := json.Marshal(tr2.process.counters)
			sm.PutStr(string(Trace2ProcessCounters), string(jargs))
		}
	}
}

func emitNonMainThreadSpan(span *ptrace.Span, th *TrThread, tr2 *trace2Dataset) {
	emitSpanEssentials(span, &th.lifetime, tr2)

	sm := span.Attributes()
	sm.PutStr(string(Trace2SpanType), "thread")

	if th.timers != nil {
		jargs, _ := json.Marshal(th.timers)
		sm.PutStr(string(Trace2ThreadTimers), string(jargs))
	}

	if th.counters != nil {
		jargs, _ := json.Marshal(th.counters)
		sm.PutStr(string(Trace2ThreadCounters), string(jargs))
	}
}

func emitRegionSpan(span *ptrace.Span, r *TrRegion, tr2 *trace2Dataset) {
	emitSpanEssentials(span, &r.lifetime, tr2)

	sm := span.Attributes()
	sm.PutStr(string(Trace2SpanType), "region")

	sm.PutStr(string(Trace2RegionRepoId), fmt.Sprintf("%d", r.repoId))

	sm.PutStr(string(Trace2RegionNesting), fmt.Sprintf("%d", r.nestingLevel))
	if len(r.message) > 0 {
		sm.PutStr(string(Trace2RegionMessage), r.message)
	}

	if r.dataValues != nil && len(r.dataValues) > 0 {
		jargs, _ := json.Marshal(r.dataValues)
		sm.PutStr(string(Trace2RegionData), string(jargs))
	}
}

func emitChildSpan(span *ptrace.Span, child *TrChild, tr2 *trace2Dataset) {
	emitSpanEssentials(span, &child.lifetime, tr2)

	sm := span.Attributes()
	sm.PutStr(string(Trace2SpanType), "child")

	if len(child.argv) > 0 {
		jargs, _ := json.Marshal(child.argv)
		sm.PutStr(string(Trace2ChildArgv), string(jargs))
	}

	// Azure automatically treats integer attributes as "customMeasurements"
	// rather than grouping them with the other "customDimensions".  Or they
	// appear in "customDimensions" with value "".  The former can lead to
	// weird graphs where data is plotted by PID. So force them to be strings.
	sm.PutStr(string(Trace2ChildPid), fmt.Sprintf("%d", child.pid))
	sm.PutStr(string(Trace2ChildExitCode), fmt.Sprintf("%d", child.exitcode))

	if len(child.readystate) > 0 {
		// This was an async child sent to background.
		sm.PutStr(string(Trace2ChildReadyState), child.readystate)
	}

	sm.PutStr(string(Trace2ChildClass), child.class)
	if child.class == "hook" {
		sm.PutStr(string(Trace2ChildHookName), child.hookname)
	}
}

func emitExecSpan(span *ptrace.Span, e *TrExec, tr2 *trace2Dataset) {
	emitSpanEssentials(span, &e.lifetime, tr2)

	sm := span.Attributes()
	sm.PutStr(string(Trace2SpanType), "exec")

	if len(e.argv) > 0 {
		jargs, _ := json.Marshal(e.argv)
		sm.PutStr(string(Trace2ExecArgv), string(jargs))
	}

	sm.PutStr(string(Trace2ExecExe), e.exe)
	sm.PutStr(string(Trace2ExecExitCode), fmt.Sprintf("%d", e.exitcode))
}