Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[exporter/elasticsearch] Add OTel mapping mode for traces #34472

Merged
merged 18 commits into from
Aug 23, 2024
Merged

[exporter/elasticsearch] Add OTel mapping mode for traces #34472

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
e7f41d2
[exporter/elasticsearch] Add OTel mapping mode for traces
carsonip Aug 8, 2024
07ac2e7
Add changelog
carsonip Aug 8, 2024
e8337c7
Handle empty resource.schema_url and status.*
carsonip Aug 9, 2024
66ee18f
Remove trace_flags
carsonip Aug 9, 2024
b2cec62
Remove trace flags for logs
carsonip Aug 12, 2024
d19eebf
Remove trace_flags from logs
carsonip Aug 12, 2024
fd5da92
Stop using deprecated span.Kind
carsonip Aug 12, 2024
9d6d3db
Merge branch 'main' into otel-mode-traces-independent
carsonip Aug 12, 2024
7c6f811
Update otel mode logs test
carsonip Aug 12, 2024
b45df76
Refactor DS keys handling in OTel mode
carsonip Aug 13, 2024
fb69873
Fix model_test
carsonip Aug 14, 2024
3130cf1
Add tracking issues
carsonip Aug 15, 2024
54925f4
Merge branch 'main' into otel-mode-traces-independent
carsonip Aug 19, 2024
71cc306
Refactor metrics otel mode
carsonip Aug 19, 2024
f6416f1
Make scope handling consistent with resource
carsonip Aug 19, 2024
9b8588e
Update changelog
carsonip Aug 19, 2024
deb70c4
Update changelog to make it breaking
carsonip Aug 20, 2024
f18b053
Update changelog
carsonip Aug 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions .chloggen/elasticsearchexporter_otel-mode-traces.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
# Use this changelog template to create an entry for release notes.

# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
change_type: enhancement

# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
component: elasticsearchexporter

# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
note: Add OTel mapping mode for traces

# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
issues: [34472]
carsonip marked this conversation as resolved.
Show resolved Hide resolved

# (Optional) One or more lines of additional information to render under the primary note.
# These lines will be padded with 2 spaces and then inserted directly into the document.
# Use pipe (|) for multiline entries.
subtext:

# If your change doesn't affect end users or the exported elements of any package,
# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
# Optional: The change log or logs in which this entry should be included.
# e.g. '[user]' or '[user, api]'
# Include 'user' if the change is relevant to end users.
# Include 'api' if there is a change to a library API.
# Default: '[user]'
change_logs: [user]
6 changes: 4 additions & 2 deletions exporter/elasticsearchexporter/exporter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,7 @@ func (e *elasticsearchExporter) pushTraceData(
spans := scopeSpan.Spans()
for k := 0; k < spans.Len(); k++ {
span := spans.At(k)
if err := e.pushTraceRecord(ctx, resource, span, scope, session); err != nil {
if err := e.pushTraceRecord(ctx, resource, il.SchemaUrl(), span, scope, scopeSpan.SchemaUrl(), session); err != nil {
if cerr := ctx.Err(); cerr != nil {
return cerr
}
Expand All @@ -366,8 +366,10 @@ func (e *elasticsearchExporter) pushTraceData(
func (e *elasticsearchExporter) pushTraceRecord(
ctx context.Context,
resource pcommon.Resource,
resourceSchemaURL string,
span ptrace.Span,
scope pcommon.InstrumentationScope,
scopeSchemaURL string,
bulkIndexerSession bulkIndexerSession,
) error {
fIndex := e.index
Expand All @@ -383,7 +385,7 @@ func (e *elasticsearchExporter) pushTraceRecord(
fIndex = formattedIndex
}

document, err := e.model.encodeSpan(resource, span, scope)
document, err := e.model.encodeSpan(resource, resourceSchemaURL, span, scope, scopeSchemaURL)
if err != nil {
return fmt.Errorf("failed to encode trace record: %w", err)
}
Expand Down
62 changes: 62 additions & 0 deletions exporter/elasticsearchexporter/exporter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -880,6 +880,68 @@ func TestExporterTraces(t *testing.T) {
))
rec.WaitItems(1)
})

t.Run("otel mode", func(t *testing.T) {
rec := newBulkRecorder()
server := newESTestServer(t, func(docs []itemRequest) ([]itemResponse, error) {
rec.Record(docs)
return itemsAllOK(docs)
})

exporter := newTestTracesExporter(t, server.URL, func(cfg *Config) {
cfg.TracesDynamicIndex.Enabled = true
cfg.Mapping.Mode = "otel"
})

traces := ptrace.NewTraces()
resourceSpans := traces.ResourceSpans()
rs := resourceSpans.AppendEmpty()

span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
span.SetName("name")
span.SetTraceID(pcommon.NewTraceIDEmpty())
span.SetSpanID(pcommon.NewSpanIDEmpty())
span.SetFlags(1)
span.SetDroppedAttributesCount(2)
span.SetDroppedEventsCount(3)
span.SetDroppedLinksCount(4)
span.TraceState().FromRaw("foo")
span.SetStartTimestamp(pcommon.NewTimestampFromTime(time.Unix(3600, 0)))
span.SetEndTimestamp(pcommon.NewTimestampFromTime(time.Unix(7200, 0)))

scopeAttr := span.Attributes()
fillResourceAttributeMap(scopeAttr, map[string]string{
"attr.foo": "attr.bar",
})

resAttr := rs.Resource().Attributes()
fillResourceAttributeMap(resAttr, map[string]string{
"resource.foo": "resource.bar",
})

spanLink := span.Links().AppendEmpty()
spanLink.SetTraceID(pcommon.NewTraceIDEmpty())
spanLink.SetSpanID(pcommon.NewSpanIDEmpty())
spanLink.SetFlags(10)
spanLink.SetDroppedAttributesCount(11)
spanLink.TraceState().FromRaw("bar")
fillResourceAttributeMap(spanLink.Attributes(), map[string]string{
"link.attr.foo": "link.attr.bar",
})

mustSendTraces(t, exporter, traces)

rec.WaitItems(1)

expected := []itemRequest{
{
Action: []byte(`{"create":{"_index":"traces-generic.otel-default"}}`),
Document: []byte(`{"@timestamp":"1970-01-01T01:00:00.000000000Z","attributes":{"attr.foo":"attr.bar"},"data_stream":{"dataset":"generic.otel","namespace":"default","type":"traces"},"dropped_attributes_count":2,"dropped_events_count":3,"dropped_links_count":4,"duration":3600000000000,"kind":"SPAN_KIND_UNSPECIFIED","links":[{"attributes":{"link.attr.foo":"link.attr.bar"},"dropped_attributes_count":11,"span_id":"","trace_flags":10,"trace_id":"","trace_state":"bar"}],"name":"name","resource":{"attributes":{"resource.foo":"resource.bar"},"dropped_attributes_count":0},"status":{"code":"Unset"},"trace_flags":1,"trace_state":"foo"}`),
},
}

assertItemsEqual(t, expected, rec.Items(), false)
})
}

// TestExporterAuth verifies that the Elasticsearch exporter supports
Expand Down
200 changes: 132 additions & 68 deletions exporter/elasticsearchexporter/model.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ var resourceAttrsToPreserve = map[string]bool{

type mappingModel interface {
encodeLog(pcommon.Resource, string, plog.LogRecord, pcommon.InstrumentationScope, string) ([]byte, error)
encodeSpan(pcommon.Resource, ptrace.Span, pcommon.InstrumentationScope) ([]byte, error)
encodeSpan(pcommon.Resource, string, ptrace.Span, pcommon.InstrumentationScope, string) ([]byte, error)
upsertMetricDataPointValue(map[uint32]objmodel.Document, pcommon.Resource, pcommon.InstrumentationScope, pmetric.Metric, dataPoint, pcommon.Value) error
encodeDocument(objmodel.Document) ([]byte, error)
}
Expand Down Expand Up @@ -130,7 +130,29 @@ func (m *encodeModel) encodeLogDefaultMode(resource pcommon.Resource, record plo
return document
}

var datastreamKeys = []string{dataStreamType, dataStreamDataset, dataStreamNamespace}
func forEachDataStreamKey(fn func(key string)) {
for _, key := range []string{dataStreamType, dataStreamDataset, dataStreamNamespace} {
fn(key)
}
}

// addDataStreamAttributes adds data_stream.* attributes to document
func addDataStreamAttributes(document *objmodel.Document, attr pcommon.Map) {
forEachDataStreamKey(func(key string) {
if value, exists := attr.Get(key); exists {
document.AddAttribute(key, value)
}
})
}

// stripDataStreamAttributes removes data_stream.* attributes from map
func stripDataStreamAttributes(attr pcommon.Map) {
forEachDataStreamKey(func(key string) {
if _, exists := attr.Get(key); exists {
attr.Remove(key)
}
})
}

func (m *encodeModel) encodeLogOTelMode(resource pcommon.Resource, resourceSchemaURL string, record plog.LogRecord, scope pcommon.InstrumentationScope, scopeSchemaURL string) objmodel.Document {
var document objmodel.Document
Expand All @@ -154,67 +176,12 @@ func (m *encodeModel) encodeLogOTelMode(resource pcommon.Resource, resourceSchem
// updated by the router.
// Move them to the top of the document and remove them from the record
attributeMap := record.Attributes()

forEachDataStreamKey := func(fn func(key string)) {
for _, key := range datastreamKeys {
fn(key)
}
}

forEachDataStreamKey(func(key string) {
if value, exists := attributeMap.Get(key); exists {
document.AddAttribute(key, value)
attributeMap.Remove(key)
}
})

addDataStreamAttributes(&document, attributeMap)
stripDataStreamAttributes(attributeMap)
Copy link
Member

@lahsivjar lahsivjar Aug 9, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: pcommon.Map#Get could get expensive with more attributes. Though I don't think we would have a huge number of keys for it to be a conecern but it is not helping with readability of the code either. Also, I am not a fan of these util functions. How about:

	attributeMap.RemoveIf(func(k string, v pcommon.Value) bool {
		// just a few keys in datastream keys
		if slices.Contains(datastreamKeys, k) {
			// Add attribute will not add empty values
			document.AddAttribute(k, v)
			return true
		}
		return false
	})

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree that it is not great. It was designed as such as we need to call addDataStreamAttributes for record level attributes, and stripDataStreamAttributes for all resource, scope, and record level attributes. Splitting them into 2 functions makes it easier to understand. Performance would have been fine if attributes Get was O(1), but the current implementation isn't.

Would you still prefer implementing it inline for all 3 kinds of attributes?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would you still prefer implementing it inline for all 3 kinds of attributes?

I think the function call is simple enough to inline and would also be better for readability. So, my vote is to inline.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done, looking much cleaner now.

document.AddAttributes("attributes", attributeMap)

// Resource
resourceMapVal := pcommon.NewValueMap()
resourceMap := resourceMapVal.Map()
resourceMap.PutStr("schema_url", resourceSchemaURL)
resourceMap.PutInt("dropped_attributes_count", int64(resource.DroppedAttributesCount()))
resourceAttrMap := resourceMap.PutEmptyMap("attributes")

resource.Attributes().CopyTo(resourceAttrMap)

// Remove data_stream attributes from the resources attributes if present
forEachDataStreamKey(func(key string) {
resourceAttrMap.Remove(key)
})

document.Add("resource", objmodel.ValueFromAttribute(resourceMapVal))

// Scope
scopeMapVal := pcommon.NewValueMap()
scopeMap := scopeMapVal.Map()
if scope.Name() != "" {
scopeMap.PutStr("name", scope.Name())
}
if scope.Version() != "" {
scopeMap.PutStr("version", scope.Version())
}
if scopeSchemaURL != "" {
scopeMap.PutStr("schema_url", scopeSchemaURL)
}
if scope.DroppedAttributesCount() > 0 {
scopeMap.PutInt("dropped_attributes_count", int64(scope.DroppedAttributesCount()))
}
scopeAttributes := scope.Attributes()
if scopeAttributes.Len() > 0 {
scopeAttrMap := scopeMap.PutEmptyMap("attributes")
scopeAttributes.CopyTo(scopeAttrMap)

// Remove data_stream attributes from the scope attributes if present
forEachDataStreamKey(func(key string) {
scopeAttrMap.Remove(key)
})
}

if scopeMap.Len() > 0 {
document.Add("scope", objmodel.ValueFromAttribute(scopeMapVal))
}
m.encodeResourceOTelMode(&document, resource, resourceSchemaURL)
m.encodeScopeOTelMode(&document, scope, scopeSchemaURL)

// Body
setOTelLogBody(&document, record.Body())
Expand Down Expand Up @@ -385,7 +352,109 @@ func numberToValue(dp pmetric.NumberDataPoint) (pcommon.Value, error) {
return pcommon.Value{}, errInvalidNumberDataPoint
}

func (m *encodeModel) encodeSpan(resource pcommon.Resource, span ptrace.Span, scope pcommon.InstrumentationScope) ([]byte, error) {
func (m *encodeModel) encodeResourceOTelMode(document *objmodel.Document, resource pcommon.Resource, resourceSchemaURL string) {
resourceMapVal := pcommon.NewValueMap()
resourceMap := resourceMapVal.Map()
if resourceSchemaURL != "" {
andrzej-stencel marked this conversation as resolved.
Show resolved Hide resolved
resourceMap.PutStr("schema_url", resourceSchemaURL)
}
resourceMap.PutInt("dropped_attributes_count", int64(resource.DroppedAttributesCount()))
resourceAttrMap := resourceMap.PutEmptyMap("attributes")
resource.Attributes().CopyTo(resourceAttrMap)
stripDataStreamAttributes(resourceAttrMap)

document.Add("resource", objmodel.ValueFromAttribute(resourceMapVal))
carsonip marked this conversation as resolved.
Show resolved Hide resolved
}

func (m *encodeModel) encodeScopeOTelMode(document *objmodel.Document, scope pcommon.InstrumentationScope, scopeSchemaURL string) {
scopeMapVal := pcommon.NewValueMap()
scopeMap := scopeMapVal.Map()
if scope.Name() != "" {
scopeMap.PutStr("name", scope.Name())
}
if scope.Version() != "" {
scopeMap.PutStr("version", scope.Version())
}
if scopeSchemaURL != "" {
scopeMap.PutStr("schema_url", scopeSchemaURL)
}
if scope.DroppedAttributesCount() > 0 {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is a difference of treatment compared to the resource level. Don't you want to set this always even if zero?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice catch. I think it was initially implemented in this manner in a previous PR for efficiency: #33290 . Also, scope is skipped entirely if there is nothing interesting, unlike resource.

But I'll get back to the team and see if there is any implication on data correctness.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated the PR to keep the handling consistent

scopeMap.PutInt("dropped_attributes_count", int64(scope.DroppedAttributesCount()))
}
scopeAttributes := scope.Attributes()
if scopeAttributes.Len() > 0 {
scopeAttrMap := scopeMap.PutEmptyMap("attributes")
scopeAttributes.CopyTo(scopeAttrMap)
stripDataStreamAttributes(scopeAttrMap)
}
if scopeMap.Len() > 0 {
document.Add("scope", objmodel.ValueFromAttribute(scopeMapVal))
}
}

func (m *encodeModel) encodeSpan(resource pcommon.Resource, resourceSchemaURL string, span ptrace.Span, scope pcommon.InstrumentationScope, scopeSchemaURL string) ([]byte, error) {
var document objmodel.Document
switch m.mode {
case MappingOTel:
document = m.encodeSpanOTelMode(resource, resourceSchemaURL, span, scope, scopeSchemaURL)
default:
document = m.encodeSpanDefaultMode(resource, span, scope)
}
document.Dedup()
var buf bytes.Buffer
err := document.Serialize(&buf, m.dedot, m.mode == MappingOTel)
return buf.Bytes(), err
}

func (m *encodeModel) encodeSpanOTelMode(resource pcommon.Resource, resourceSchemaURL string, span ptrace.Span, scope pcommon.InstrumentationScope, scopeSchemaURL string) objmodel.Document {
var document objmodel.Document
document.AddTimestamp("@timestamp", span.StartTimestamp())
document.AddTraceID("trace_id", span.TraceID())
document.AddSpanID("span_id", span.SpanID())
document.AddString("trace_state", span.TraceState().AsRaw())
document.AddSpanID("parent_span_id", span.ParentSpanID())
carsonip marked this conversation as resolved.
Show resolved Hide resolved
document.AddInt("trace_flags", int64(span.Flags()))
carsonip marked this conversation as resolved.
Show resolved Hide resolved
document.AddString("name", span.Name())
document.AddString("kind", traceutil.SpanKindStr(span.Kind()))
document.AddInt("duration", int64(span.EndTimestamp()-span.StartTimestamp()))

attributeMap := span.Attributes()
addDataStreamAttributes(&document, attributeMap)
stripDataStreamAttributes(attributeMap)
document.AddAttributes("attributes", attributeMap)

document.AddInt("dropped_attributes_count", int64(span.DroppedAttributesCount()))
document.AddInt("dropped_events_count", int64(span.DroppedEventsCount()))

links := pcommon.NewValueSlice()
linkSlice := links.SetEmptySlice()
spanLinks := span.Links()
for i := 0; i < spanLinks.Len(); i++ {
linkMap := linkSlice.AppendEmpty().SetEmptyMap()
spanLink := spanLinks.At(i)
linkMap.PutStr("trace_id", spanLink.TraceID().String())
linkMap.PutStr("span_id", spanLink.SpanID().String())
linkMap.PutStr("trace_state", spanLink.TraceState().AsRaw())
mAttr := linkMap.PutEmptyMap("attributes")
spanLink.Attributes().CopyTo(mAttr)
linkMap.PutInt("dropped_attributes_count", int64(spanLink.DroppedAttributesCount()))
linkMap.PutInt("trace_flags", int64(spanLink.Flags()))
carsonip marked this conversation as resolved.
Show resolved Hide resolved
}
document.AddAttribute("links", links)

document.AddInt("dropped_links_count", int64(span.DroppedLinksCount()))
document.AddString("status.message", span.Status().Message())
document.AddString("status.code", span.Status().Code().String())

m.encodeResourceOTelMode(&document, resource, resourceSchemaURL)
m.encodeScopeOTelMode(&document, scope, scopeSchemaURL)

// TODO: add span events to log data streams
andrzej-stencel marked this conversation as resolved.
Show resolved Hide resolved

return document
}

func (m *encodeModel) encodeSpanDefaultMode(resource pcommon.Resource, span ptrace.Span, scope pcommon.InstrumentationScope) objmodel.Document {
var document objmodel.Document
document.AddTimestamp("@timestamp", span.StartTimestamp()) // We use @timestamp in order to ensure that we can index if the default data stream logs template is used.
document.AddTimestamp("EndTimestamp", span.EndTimestamp())
Expand All @@ -402,12 +471,7 @@ func (m *encodeModel) encodeSpan(resource pcommon.Resource, span ptrace.Span, sc
m.encodeEvents(&document, span.Events())
document.AddInt("Duration", durationAsMicroseconds(span.StartTimestamp().AsTime(), span.EndTimestamp().AsTime())) // unit is microseconds
document.AddAttributes("Scope", scopeToAttributes(scope))
document.Dedup()

var buf bytes.Buffer
// OTel serialization is not supported for traces yet
err := document.Serialize(&buf, m.dedot, false)
return buf.Bytes(), err
return document
}

func (m *encodeModel) encodeAttributes(document *objmodel.Document, attributes pcommon.Map) {
Expand Down
2 changes: 1 addition & 1 deletion exporter/elasticsearchexporter/model_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ var expectedLogBodyDeDottedWithEmptyTimestamp = `{"@timestamp":"1970-01-01T00:00
func TestEncodeSpan(t *testing.T) {
model := &encodeModel{dedot: false}
td := mockResourceSpans()
spanByte, err := model.encodeSpan(td.ResourceSpans().At(0).Resource(), td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0), td.ResourceSpans().At(0).ScopeSpans().At(0).Scope())
spanByte, err := model.encodeSpan(td.ResourceSpans().At(0).Resource(), "", td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0), td.ResourceSpans().At(0).ScopeSpans().At(0).Scope(), "")
assert.NoError(t, err)
assert.Equal(t, expectedSpanBody, string(spanByte))
}
Expand Down
Loading