diff --git a/.chloggen/47465-opensearch-dotted-attribute-conflict.yaml b/.chloggen/47465-opensearch-dotted-attribute-conflict.yaml new file mode 100644 index 0000000000000..8d1208591e14f --- /dev/null +++ b/.chloggen/47465-opensearch-dotted-attribute-conflict.yaml @@ -0,0 +1,36 @@ +# Use this changelog template to create an entry for release notes. + +# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' +change_type: bug_fix + +# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog) +component: exporter/opensearch + +# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). +note: Avoid OpenSearch mapping conflicts (mapper_parsing_exception) when an attribute is used both as a value and as an object prefix. + +# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. +issues: [47465] + +# (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: | + OpenSearch expands dots in JSON field names into nested objects during dynamic + mapping, so a flat OTel attribute map containing both a concrete key (e.g. + `code.function`) and a longer key that uses it as an object prefix (e.g. + `code.function.name`) was rejected. The exporter now relocates the conflicting + concrete value under a `.value` sub-key (`code.function` -> `code.function.value`) + so the document is indexable. This applies to span, event, link, log-record, + metric data-point, exemplar and instrumentation-scope attributes in the ss4o + and otel-v1 mapping modes. Only documents that would otherwise be rejected are + changed. + +# 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] diff --git a/.chloggen/47465-opensearch-ss4o-manage-index-template.yaml b/.chloggen/47465-opensearch-ss4o-manage-index-template.yaml new file mode 100644 index 0000000000000..1e1d5e35a882b --- /dev/null +++ b/.chloggen/47465-opensearch-ss4o-manage-index-template.yaml @@ -0,0 +1,35 @@ +# 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. receiver/filelog) +component: exporter/opensearch + +# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). +note: Support `mapping.manage_index_template` in the `ss4o` mapping mode to install index templates that map attribute bags as `flat_object`. + +# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. +issues: [47465] + +# (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: | + Previously `manage_index_template` was only valid with `otel-v1`. In `ss4o` + mode the exporter now installs traces and logs index templates that map the + attribute-bearing objects as `flat_object`, so OpenSearch stops expanding dots + in attribute keys into nested objects. This deterministically prevents + dotted-attribute mapping conflicts across documents. Note that `flat_object` + indexes all values as strings, so type-aware (e.g. numeric range) queries on + attribute values are not available; the option is opt-in and defaults to + `false`. + +# 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] diff --git a/exporter/opensearchexporter/README.md b/exporter/opensearchexporter/README.md index 1561f441057fd..3efbd1fb2efd9 100644 --- a/exporter/opensearchexporter/README.md +++ b/exporter/opensearchexporter/README.md @@ -117,7 +117,7 @@ The OpenSearch exporter supports several document schemas and preprocessing beha - `flatten_attributes`: Uses the ECS mapping but flattens all resource and log attributes in the record to the top-level. - `bodymap`: uses the "body" of a log record as the exact content of the OpenSearch document, without any transformation. This mapping mode is intended for use cases where the client wishes to have complete control over the OpenSearch document structure. - `otel-v1`: exports logs and traces using the Data Prepper OTel v1 schema, compatible with OpenSearch Observability dashboards. - - `manage_index_template`: (optional, default=`false`) When `true`, creates composable index templates on startup. Only valid with `otel-v1` mode. + - `manage_index_template`: (optional, default=`false`) When `true`, creates composable index templates on startup. Only valid with `otel-v1` and `ss4o` modes. In `ss4o` mode the templates map attribute bags as `flat_object` to avoid dotted-attribute mapping conflicts (see [SS4O mapping mode](#ss4o-mapping-mode)). - `timestamp_field`: (optional) Field to store the timestamp in. If not set, uses the default `@timestamp`. - `unix_timestamp`: (optional) Whether to store the timestamp in epoch milliseconds. - `dedup`: (optional) removes fields from the document, that have duplicate keys. The filtering only keeps the last value for a key. @@ -142,6 +142,18 @@ Metrics follow the [SS4O metrics schema](https://github.com/opensearch-project/o > > The catalog template maps `value@int` as a 32-bit `integer`; OpenTelemetry integer data points are 64-bit, so values beyond ±2³¹ are rejected. Change that field to `long` in the template before installing it ([Data Prepper's reference template](https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/resources/index-template/metrics-otel-v1-index-standard-template.json) already uses `long`). +##### Dotted attribute keys and mapping conflicts + +OpenTelemetry attributes are a flat namespace in which `code.function` and `code.function.name` are two distinct keys. OpenSearch, however, expands dots in JSON field names into nested objects during dynamic mapping. When both keys are present, OpenSearch tries to map `attributes.code.function` as *both* a concrete value and an object and rejects the document with a `mapper_parsing_exception`. This commonly happens while migrating between semantic-convention versions (e.g. `code.function` → `code.function.name`). + +The exporter handles this in two complementary ways: + +- **Within a document (always on):** if an attribute map contains both a concrete key and a longer key that uses it as an object prefix, the concrete value is relocated under a `.value` sub-key (`code.function` → `code.function.value`) so the document is indexable. This applies to span, event, link, log-record, metric data-point, exemplar and instrumentation-scope attributes (in the `ss4o` and `otel-v1` modes), and only changes documents OpenSearch would otherwise reject. +- **Across documents (opt-in):** set `manage_index_template: true` to have the exporter install traces and logs index templates that map the attribute-bearing objects (`attributes`, `resource`, scope/event/link attributes) as [`flat_object`](https://docs.opensearch.org/latest/mappings/supported-field-types/flat-object/). OpenSearch then stores attribute bags without expanding dots, which prevents conflicts even when the conflicting keys arrive in separate documents. + +> [!NOTE] +> `flat_object` indexes every leaf value as a string, so type-aware queries (for example numeric range queries) on attribute values are not available. The template only matches the default `ss4o_traces-*` / `ss4o_logs-*` index patterns; if you configure a custom `traces_index`/`logs_index`, install an equivalent template out-of-band. + #### ECS mapping mode > [!WARNING] diff --git a/exporter/opensearchexporter/config.go b/exporter/opensearchexporter/config.go index 56b3b3ae4ac25..e243f64952615 100644 --- a/exporter/opensearchexporter/config.go +++ b/exporter/opensearchexporter/config.go @@ -82,7 +82,7 @@ var ( errMetricsIndexTimeFormatInvalid = errors.New("metrics_index_time_format contains unsupported or invalid tokens") errOTelV1DatasetNamespaceUnused = errors.New(`dataset and namespace are not used by mapping.mode "otel-v1"; remove them or pick a different mode`) errMetricsMappingModeUnsupported = errors.New(`metrics are only supported by mapping.mode "ss4o" and "otel-v1"`) - errManageIndexTemplateInvalidMode = errors.New("mapping.manage_index_template is only supported with mapping.mode \"otel-v1\"") + errManageIndexTemplateInvalidMode = errors.New("mapping.manage_index_template is only supported with mapping.mode \"otel-v1\" or \"ss4o\"") ) type MappingsSettings struct { @@ -114,7 +114,15 @@ type MappingsSettings struct { Mode string `mapstructure:"mode"` // ManageIndexTemplate controls whether the exporter creates index templates on startup. - // Only supported when Mode is "otel-v1". Validation will reject this option with other modes. + // Only supported when Mode is "otel-v1" or "ss4o". Validation will reject this option + // with other modes. + // + // In "ss4o" mode the templates map attribute bags (attributes, resource and scope/event/ + // link attributes) as flat_object so OpenSearch does not expand dots in attribute keys + // into nested objects, preventing mapper_parsing_exception conflicts between an attribute + // used as a value ("code.function") and the same prefix used as an object + // ("code.function.name"). Note that flat_object indexes all values as strings, so + // type-aware (e.g. numeric range) queries on attribute values are not available. ManageIndexTemplate bool `mapstructure:"manage_index_template"` // Additional field mappings. @@ -234,7 +242,9 @@ func (cfg *Config) Validate() error { multiErr = append(multiErr, errMappingModeInvalid) } - if cfg.MappingsSettings.ManageIndexTemplate && cfg.MappingsSettings.Mode != MappingOTelV1.String() { + if cfg.MappingsSettings.ManageIndexTemplate && + cfg.MappingsSettings.Mode != MappingOTelV1.String() && + cfg.MappingsSettings.Mode != MappingSS4O.String() { multiErr = append(multiErr, errManageIndexTemplateInvalidMode) } diff --git a/exporter/opensearchexporter/config.schema.yaml b/exporter/opensearchexporter/config.schema.yaml index 19ea6e7315122..3148c163a549c 100644 --- a/exporter/opensearchexporter/config.schema.yaml +++ b/exporter/opensearchexporter/config.schema.yaml @@ -16,7 +16,7 @@ $defs: description: File to read additional fields mappings from. type: string manage_index_template: - description: ManageIndexTemplate controls whether the exporter creates index templates on startup. Only supported when Mode is "otel-v1". Validation will reject this option with other modes. + description: 'ManageIndexTemplate controls whether the exporter creates index templates on startup. Only supported when Mode is "otel-v1" or "ss4o". Validation will reject this option with other modes. In "ss4o" mode the templates map attribute bags as flat_object so OpenSearch does not expand dots in attribute keys into nested objects, preventing mapper_parsing_exception conflicts; note that flat_object indexes all values as strings, so type-aware queries on attribute values are not available.' type: boolean mode: description: 'Mode configures the field mappings. Supported modes are the following: ss4o: exports logs in the Simple Schema for Observability standard. This mode is enabled by default. See: https://opensearch.org/docs/latest/observing-your-data/ss4o/ ecs: maps fields defined in the OpenTelemetry Semantic Conventions to the Elastic Common Schema. See: https://www.elastic.co/guide/en/ecs/current/index.html flatten_attributes: uses the ECS mapping but flattens all resource and log attributes in the record to the top-level. bodymap: supports only logs and uses the "body" of a log record as the exact content of the OpenSearch document, without any transformation. This mapping mode is intended for use cases where the client wishes to have complete control over the OpenSearch document structure. otel-v1: exports logs and traces using the OTel v1 schema published by the OpenSearch Data Prepper project (https://github.com/opensearch-project/data-prepper). Documents are compatible with OpenSearch Observability dashboards that consume Data Prepper indices. See the upstream index templates for the canonical field mappings: https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/resources/index-template/otel-v1-apm-span-index-standard-template.json https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/resources/index-template/logs-otel-v1-index-standard-template.json' diff --git a/exporter/opensearchexporter/config_test.go b/exporter/opensearchexporter/config_test.go index cb2a3e18176c0..1116ae6e4a559 100644 --- a/exporter/opensearchexporter/config_test.go +++ b/exporter/opensearchexporter/config_test.go @@ -354,15 +354,20 @@ func TestOTelV1MappingModeValidation(t *testing.T) { manageTpl: true, }, { - name: "ss4o with manage_index_template true is invalid", - mode: "ss4o", - manageTpl: true, - expectError: errManageIndexTemplateInvalidMode.Error(), + name: "ss4o with manage_index_template true is valid", + mode: "ss4o", + manageTpl: true, }, { name: "ss4o with manage_index_template false is valid", mode: "ss4o", }, + { + name: "ecs with manage_index_template true is invalid", + mode: "ecs", + manageTpl: true, + expectError: errManageIndexTemplateInvalidMode.Error(), + }, } for _, tt := range tests { diff --git a/exporter/opensearchexporter/encoder.go b/exporter/opensearchexporter/encoder.go index 0933e11ed670b..3c14c1b7d0844 100644 --- a/exporter/opensearchexporter/encoder.go +++ b/exporter/opensearchexporter/encoder.go @@ -8,6 +8,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "time" "go.opentelemetry.io/collector/pdata/pcommon" @@ -22,6 +24,80 @@ import ( var errInvalidTypeForBodyMapMode = errors.New("invalid log record body type for 'bodymap' mapping mode") +// resolveAttributeKeyConflicts rewrites attribute keys that would otherwise +// cause an OpenSearch mapping conflict. +// +// OpenSearch expands dots in JSON field names into nested objects during +// dynamic mapping. A flat OTel attribute map that contains both a concrete key +// (e.g. "code.function") and a longer key that uses it as an object prefix +// (e.g. "code.function.name") makes OpenSearch try to map +// "attributes.code.function" as both a concrete value and an object, which it +// rejects with a mapper_parsing_exception. This is common while migrating +// between semantic-convention versions (code.function -> code.function.name). +// +// To keep the document indexable, the concrete value is moved under a ".value" +// sub-key ("code.function" -> "code.function.value"), mirroring the behaviour +// of the ECS mapping mode's objmodel.Dedup step. The rewrite only triggers when +// a conflicting sibling is present in the same map, i.e. only for documents +// OpenSearch would otherwise reject, so well-formed documents are unchanged. +// It recurses into nested maps and arrays of maps so conflicts within map-typed +// attribute values are handled too. +func resolveAttributeKeyConflicts(m map[string]any) { + if len(m) == 0 { + return + } + + // Handle nested maps and arrays of maps first. + for _, v := range m { + switch vv := v.(type) { + case map[string]any: + resolveAttributeKeyConflicts(vv) + case []any: + for _, e := range vv { + if em, ok := e.(map[string]any); ok { + resolveAttributeKeyConflicts(em) + } + } + } + } + + // Repeatedly rename the shortest conflicting key until the map is stable. + // Renaming can, in pathological cases, create a new adjacency, so re-scan + // with a fresh, sorted key set after each rename. + for { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + + renamed := false + for i := 0; i < len(keys)-1; i++ { + key, next := keys[i], keys[i+1] + // next must use key as a strict, dot-delimited object prefix. + if len(key) >= len(next) || !strings.HasPrefix(next, key) || next[len(key)] != '.' { + continue + } + // Only a concrete (non-object) value conflicts with the prefix use. + if _, isObj := m[key].(map[string]any); isObj { + continue + } + target := key + ".value" + if _, exists := m[target]; !exists { + m[target] = m[key] + } + // If target already exists it is an object being built from other + // keys; drop the concrete value rather than clobber it. + delete(m, key) + renamed = true + break + } + if !renamed { + return + } + } +} + type mappingModel interface { encodeLog(resource pcommon.Resource, scope pcommon.InstrumentationScope, @@ -130,6 +206,7 @@ func (m *encodeModel) encodeLogSSO( ) ([]byte, error) { sso := ssoRecord{} sso.Attributes = record.Attributes().AsRaw() + resolveAttributeKeyConflicts(sso.Attributes) sso.Body = record.Body().AsString() sso.EventName = record.EventName() @@ -161,6 +238,7 @@ func (m *encodeModel) encodeLogSSO( sso.InstrumentationScope.Version = scope.Version() sso.InstrumentationScope.SchemaURL = schemaURL sso.InstrumentationScope.Attributes = scope.Attributes().AsRaw() + resolveAttributeKeyConflicts(sso.InstrumentationScope.Attributes) sso.Severity.Text = record.SeverityText() sso.Severity.Number = int64(record.SeverityNumber()) @@ -226,6 +304,7 @@ func (m *encodeModel) encodeTrace( sso := ssoSpan{} sso.Attributes = span.Attributes().AsRaw() + resolveAttributeKeyConflicts(sso.Attributes) sso.DroppedAttributesCount = span.DroppedAttributesCount() sso.DroppedEventsCount = span.DroppedEventsCount() sso.DroppedLinksCount = span.DroppedLinksCount() @@ -247,6 +326,7 @@ func (m *encodeModel) encodeTrace( e := span.Events().At(i) ssoEvent := &sso.Events[i] ssoEvent.Attributes = e.Attributes().AsRaw() + resolveAttributeKeyConflicts(ssoEvent.Attributes) ssoEvent.DroppedAttributesCount = e.DroppedAttributesCount() ssoEvent.Name = e.Name() ts := e.Timestamp().AsTime() @@ -278,6 +358,7 @@ func (m *encodeModel) encodeTrace( sso.InstrumentationScope.Version = scope.Version() sso.InstrumentationScope.SchemaURL = schemaURL sso.InstrumentationScope.Attributes = scope.Attributes().AsRaw() + resolveAttributeKeyConflicts(sso.InstrumentationScope.Attributes) if span.Links().Len() > 0 { sso.Links = make([]ssoSpanLinks, span.Links().Len()) @@ -285,6 +366,7 @@ func (m *encodeModel) encodeTrace( link := span.Links().At(i) ssoLink := &sso.Links[i] ssoLink.Attributes = link.Attributes().AsRaw() + resolveAttributeKeyConflicts(ssoLink.Attributes) ssoLink.DroppedAttributesCount = link.DroppedAttributesCount() ssoLink.TraceID = link.TraceID().String() ssoLink.TraceState = link.TraceState().AsRaw() @@ -330,6 +412,8 @@ func (*encodeModel) encodeLogOTelV1( DroppedAttributesCount: scope.DroppedAttributesCount(), }, } + resolveAttributeKeyConflicts(doc.Attributes) + resolveAttributeKeyConflicts(doc.InstrumentationScope.Attributes) return json.Marshal(doc) } @@ -378,6 +462,8 @@ func (*encodeModel) encodeTraceOTelV1( DroppedAttributesCount: scope.DroppedAttributesCount(), }, } + resolveAttributeKeyConflicts(doc.Attributes) + resolveAttributeKeyConflicts(doc.InstrumentationScope.Attributes) // Extract serviceName from resource attributes if sn, ok := resource.Attributes().Get("service.name"); ok { @@ -405,6 +491,7 @@ func (*encodeModel) encodeTraceOTelV1( DroppedAttributesCount: e.DroppedAttributesCount(), Time: e.Timestamp().AsTime(), } + resolveAttributeKeyConflicts(doc.Events[i].Attributes) } } @@ -420,6 +507,7 @@ func (*encodeModel) encodeTraceOTelV1( Attributes: l.Attributes().AsRaw(), DroppedAttributesCount: l.DroppedAttributesCount(), } + resolveAttributeKeyConflicts(doc.Links[i].Attributes) } } diff --git a/exporter/opensearchexporter/encoder_test.go b/exporter/opensearchexporter/encoder_test.go index 23fbaf25ca81c..43886d652e0e7 100644 --- a/exporter/opensearchexporter/encoder_test.go +++ b/exporter/opensearchexporter/encoder_test.go @@ -592,3 +592,154 @@ func TestOTelV1_EncodeTrace_NonRootSpan(t *testing.T) { // Verify durationInNanos = 1 second assert.Equal(t, float64(1000000000), doc["durationInNanos"]) } + +func TestResolveAttributeKeyConflicts(t *testing.T) { + tests := []struct { + name string + input map[string]any + expected map[string]any + }{ + { + name: "no conflict", + input: map[string]any{"code.function": "doWork", "code.line.number": int64(10)}, + expected: map[string]any{"code.function": "doWork", "code.line.number": int64(10)}, + }, + { + name: "concrete value conflicts with object prefix", + input: map[string]any{"code.function": "doWork", "code.function.name": "doWork"}, + expected: map[string]any{"code.function.value": "doWork", "code.function.name": "doWork"}, + }, + { + name: "object prefix appears before concrete value", + input: map[string]any{"a.b.c": int64(1), "a.b": "x"}, + expected: map[string]any{"a.b.c": int64(1), "a.b.value": "x"}, + }, + { + name: "similar prefix that is not a path boundary is untouched", + input: map[string]any{"code.function": "doWork", "code.functionName": "doWork"}, + expected: map[string]any{"code.function": "doWork", "code.functionName": "doWork"}, + }, + { + name: "value target already present drops concrete value", + input: map[string]any{"code.function": "concrete", "code.function.value": "kept", "code.function.name": "n"}, + expected: map[string]any{"code.function.value": "kept", "code.function.name": "n"}, + }, + { + name: "nested map conflict is resolved recursively", + input: map[string]any{"outer": map[string]any{"x": "v", "x.y": "w"}}, + expected: map[string]any{"outer": map[string]any{"x.value": "v", "x.y": "w"}}, + }, + { + name: "conflict inside array of maps is resolved", + input: map[string]any{"items": []any{ + map[string]any{"p": "v", "p.q": "w"}, + }}, + expected: map[string]any{"items": []any{ + map[string]any{"p.value": "v", "p.q": "w"}, + }}, + }, + { + name: "empty map", + input: map[string]any{}, + expected: map[string]any{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resolveAttributeKeyConflicts(tt.input) + assert.Equal(t, tt.expected, tt.input) + }) + } +} + +// TestSSO_EncodeTrace_DottedAttributeConflict reproduces +// https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/47465: +// a span carrying both the old and new code.function semantic conventions must +// not emit a document that OpenSearch would reject with a mapper_parsing_exception. +func TestSSO_EncodeTrace_DottedAttributeConflict(t *testing.T) { + model := &encodeModel{dataset: "default", namespace: "namespace"} + + traces := ptrace.NewTraces() + span := traces.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans().AppendEmpty() + span.SetName("test-span") + span.Attributes().PutStr("code.function", "doWork") + span.Attributes().PutStr("code.function.name", "doWork") + + resource := pcommon.NewResource() + scope := pcommon.NewInstrumentationScope() + + result, err := model.encodeTrace(resource, scope, "", span) + require.NoError(t, err) + + var doc map[string]any + require.NoError(t, json.Unmarshal(result, &doc)) + + attrs := doc["attributes"].(map[string]any) + // The concrete value is relocated so code.function can be an object holding + // both children; there is no bare concrete "code.function" key left. + _, hasConcrete := attrs["code.function"] + assert.False(t, hasConcrete, "bare concrete code.function must be rewritten") + assert.Equal(t, "doWork", attrs["code.function.value"]) + assert.Equal(t, "doWork", attrs["code.function.name"]) +} + +// TestSSO_EncodeLog_DottedAttributeConflict verifies the log SSO path applies +// the same conflict resolution as traces. +func TestSSO_EncodeLog_DottedAttributeConflict(t *testing.T) { + model := &encodeModel{sso: true, dataset: "default", namespace: "namespace"} + + logs := plog.NewLogs() + rl := logs.ResourceLogs().AppendEmpty() + lr := rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty() + lr.Attributes().PutStr("code.function", "doWork") + lr.Attributes().PutStr("code.function.name", "doWork") + + result, err := model.encodeLog(rl.Resource(), rl.ScopeLogs().At(0).Scope(), "", lr) + require.NoError(t, err) + + var doc map[string]any + require.NoError(t, json.Unmarshal(result, &doc)) + + attrs := doc["attributes"].(map[string]any) + _, hasConcrete := attrs["code.function"] + assert.False(t, hasConcrete, "bare concrete code.function must be rewritten") + assert.Equal(t, "doWork", attrs["code.function.value"]) + assert.Equal(t, "doWork", attrs["code.function.name"]) +} + +// TestOTelV1_EncodeTrace_DottedAttributeConflict verifies the otel-v1 trace +// path resolves dotted-key conflicts for span, event and link attributes. +func TestOTelV1_EncodeTrace_DottedAttributeConflict(t *testing.T) { + model := &encodeModel{otelV1: true} + + traces := ptrace.NewTraces() + span := traces.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans().AppendEmpty() + span.SetName("test-span") + span.Attributes().PutStr("code.function", "doWork") + span.Attributes().PutStr("code.function.name", "doWork") + + ev := span.Events().AppendEmpty() + ev.Attributes().PutStr("code.function", "onEvent") + ev.Attributes().PutStr("code.function.name", "onEvent") + + link := span.Links().AppendEmpty() + link.Attributes().PutStr("code.function", "onLink") + link.Attributes().PutStr("code.function.name", "onLink") + + result, err := model.encodeTrace(pcommon.NewResource(), pcommon.NewInstrumentationScope(), "", span) + require.NoError(t, err) + + var doc map[string]any + require.NoError(t, json.Unmarshal(result, &doc)) + + assertResolved := func(attrs map[string]any) { + _, hasConcrete := attrs["code.function"] + assert.False(t, hasConcrete, "bare concrete code.function must be rewritten") + assert.Equal(t, attrs["code.function.name"], attrs["code.function.value"]) + } + + assertResolved(doc["attributes"].(map[string]any)) + assertResolved(doc["events"].([]any)[0].(map[string]any)["attributes"].(map[string]any)) + assertResolved(doc["links"].([]any)[0].(map[string]any)["attributes"].(map[string]any)) +} diff --git a/exporter/opensearchexporter/index_template_manager.go b/exporter/opensearchexporter/index_template_manager.go index b70bf21bd8dd8..3e8ccd4643640 100644 --- a/exporter/opensearchexporter/index_template_manager.go +++ b/exporter/opensearchexporter/index_template_manager.go @@ -16,6 +16,8 @@ import ( const ( otelV1SpanTemplateName = "otel-v1-apm-span-index-template" otelV1LogsTemplateName = "otel-v1-logs-index-template" + ss4oTracesTemplateName = "ss4o-traces-index-template" + ss4oLogsTemplateName = "ss4o-logs-index-template" ) type templateManager struct { @@ -30,13 +32,23 @@ func newTemplateManager(client *opensearchapi.Client, logger *zap.Logger) *templ // ensureTemplates is best-effort: it logs and returns on transient cluster // errors rather than failing the exporter Start(). A failure means OpenSearch's // dynamic mapping will pick up types from the first indexed document -// (date instead of date_nanos for timestamps); existing documents are -// unaffected. This matches the Data Prepper sink's posture for the same -// operation, which logs and retries on IOException rather than blocking -// pipeline initialization. -func (tm *templateManager) ensureTemplates(ctx context.Context) { - tm.ensureTemplate(ctx, otelV1SpanTemplateName, templates.OtelV1APMSpan) - tm.ensureTemplate(ctx, otelV1LogsTemplateName, templates.OtelV1Logs) +// (date instead of date_nanos for timestamps in otel-v1 mode; expanded dotted +// attribute keys in ss4o mode); existing documents are unaffected. This matches +// the Data Prepper sink's posture for the same operation, which logs and retries +// on IOException rather than blocking pipeline initialization. +// +// The templates installed depend on the configured mapping mode: otel-v1 mode +// installs the Data Prepper-compatible templates, ss4o mode installs templates +// that map attribute bags as flat_object to avoid dotted-key mapping conflicts. +func (tm *templateManager) ensureTemplates(ctx context.Context, mode string) { + switch mode { + case MappingOTelV1.String(): + tm.ensureTemplate(ctx, otelV1SpanTemplateName, templates.OtelV1APMSpan) + tm.ensureTemplate(ctx, otelV1LogsTemplateName, templates.OtelV1Logs) + case MappingSS4O.String(): + tm.ensureTemplate(ctx, ss4oTracesTemplateName, templates.SS4OTraces) + tm.ensureTemplate(ctx, ss4oLogsTemplateName, templates.SS4OLogs) + } } func (tm *templateManager) ensureTemplate(ctx context.Context, name, body string) { diff --git a/exporter/opensearchexporter/internal/templates/embed.go b/exporter/opensearchexporter/internal/templates/embed.go index e3c89b3c02970..5829095f65314 100644 --- a/exporter/opensearchexporter/internal/templates/embed.go +++ b/exporter/opensearchexporter/internal/templates/embed.go @@ -2,10 +2,19 @@ // SPDX-License-Identifier: Apache-2.0 // Package templates embeds the composable index templates the exporter installs -// for the otel-v1 mapping mode. The templates mirror the Data Prepper schemas -// at https://github.com/opensearch-project/data-prepper/tree/main/data-prepper-plugins/opensearch/src/main/resources/index-template +// when mapping.manage_index_template is enabled. +// +// The otel-v1 templates mirror the Data Prepper schemas at +// https://github.com/opensearch-project/data-prepper/tree/main/data-prepper-plugins/opensearch/src/main/resources/index-template // and ensure date_nanos timestamps and typed dynamic-attribute mappings before // any documents are indexed. +// +// The ss4o templates map the attribute-bearing objects as flat_object so +// OpenSearch does not expand dots in attribute keys into nested objects. This +// prevents mapper_parsing_exception mapping conflicts between an attribute used +// as a concrete value ("code.function") and the same prefix used as an object +// ("code.function.name"). See: +// https://docs.opensearch.org/latest/mappings/supported-field-types/flat-object/ package templates // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter/internal/templates" import _ "embed" @@ -19,3 +28,13 @@ var OtelV1APMSpan string // //go:embed otel-v1-logs.json var OtelV1Logs string + +// SS4OTraces is the composable index template body for traces in ss4o mode. +// +//go:embed ss4o-traces.json +var SS4OTraces string + +// SS4OLogs is the composable index template body for logs in ss4o mode. +// +//go:embed ss4o-logs.json +var SS4OLogs string diff --git a/exporter/opensearchexporter/internal/templates/ss4o-logs.json b/exporter/opensearchexporter/internal/templates/ss4o-logs.json new file mode 100644 index 0000000000000..32b3decc5e8c1 --- /dev/null +++ b/exporter/opensearchexporter/internal/templates/ss4o-logs.json @@ -0,0 +1,16 @@ +{ + "index_patterns": ["ss4o_logs-*"], + "template": { + "mappings": { + "properties": { + "attributes": { "type": "flat_object" }, + "resource": { "type": "flat_object" }, + "instrumentationScope": { + "properties": { + "attributes": { "type": "flat_object" } + } + } + } + } + } +} diff --git a/exporter/opensearchexporter/internal/templates/ss4o-traces.json b/exporter/opensearchexporter/internal/templates/ss4o-traces.json new file mode 100644 index 0000000000000..3b1655e807b69 --- /dev/null +++ b/exporter/opensearchexporter/internal/templates/ss4o-traces.json @@ -0,0 +1,26 @@ +{ + "index_patterns": ["ss4o_traces-*"], + "template": { + "mappings": { + "properties": { + "attributes": { "type": "flat_object" }, + "resource": { "type": "flat_object" }, + "instrumentationScope": { + "properties": { + "attributes": { "type": "flat_object" } + } + }, + "events": { + "properties": { + "attributes": { "type": "flat_object" } + } + }, + "links": { + "properties": { + "attributes": { "type": "flat_object" } + } + } + } + } + } +} diff --git a/exporter/opensearchexporter/metrics_encoder.go b/exporter/opensearchexporter/metrics_encoder.go index 86f94b283ee58..5750b31e129da 100644 --- a/exporter/opensearchexporter/metrics_encoder.go +++ b/exporter/opensearchexporter/metrics_encoder.go @@ -80,6 +80,7 @@ func (m *encodeModel) encodeMetricSSO( sso.InstrumentationScope.Version = scope.Version() sso.InstrumentationScope.SchemaURL = schemaURL sso.InstrumentationScope.Attributes = scope.Attributes().AsRaw() + resolveAttributeKeyConflicts(sso.InstrumentationScope.Attributes) return json.Marshal(sso) } @@ -125,6 +126,7 @@ func (*encodeModel) encodeMetricOTelV1( Attributes: scope.Attributes().AsRaw(), DroppedAttributesCount: scope.DroppedAttributesCount(), } + resolveAttributeKeyConflicts(doc.InstrumentationScope.Attributes) if serviceName, ok := resource.Attributes().Get("service.name"); ok { doc.ServiceName = serviceName.AsString() } @@ -141,6 +143,7 @@ func populateMetricDocBase(doc *metricDocBase, metric pmetric.Metric, dp metricD doc.StartTime = dp.StartTimestamp().AsTime() doc.Timestamp = dp.Timestamp().AsTime() doc.Attributes = dp.Attributes().AsRaw() + resolveAttributeKeyConflicts(doc.Attributes) switch dp := dp.(type) { case pmetric.NumberDataPoint: @@ -373,6 +376,7 @@ func makeExemplars(exemplars pmetric.ExemplarSlice) []metricExemplar { Time: e.Timestamp().AsTime(), Attributes: e.FilteredAttributes().AsRaw(), } + resolveAttributeKeyConflicts(exemplar.Attributes) switch e.ValueType() { case pmetric.ExemplarValueTypeInt: value := float64(e.IntValue()) diff --git a/exporter/opensearchexporter/metrics_encoder_test.go b/exporter/opensearchexporter/metrics_encoder_test.go index 96d93168a8aa2..bc6f65e2c5507 100644 --- a/exporter/opensearchexporter/metrics_encoder_test.go +++ b/exporter/opensearchexporter/metrics_encoder_test.go @@ -491,3 +491,38 @@ func TestTemporalityString(t *testing.T) { assert.Equal(t, "AGGREGATION_TEMPORALITY_CUMULATIVE", temporalityString(pmetric.AggregationTemporalityCumulative)) assert.Equal(t, "AGGREGATION_TEMPORALITY_UNSPECIFIED", temporalityString(pmetric.AggregationTemporalityUnspecified)) } + +// TestEncodeMetric_DottedAttributeConflict ensures the metric data-point +// attributes go through the dotted-key conflict resolution so a metric carrying +// both code.function and code.function.name does not produce a document that +// OpenSearch would reject with a mapper_parsing_exception. +func TestEncodeMetric_DottedAttributeConflict(t *testing.T) { + for _, mode := range []struct { + name string + model *encodeModel + }{ + {name: "ss4o", model: &encodeModel{sso: true, dataset: "default", namespace: "namespace"}}, + {name: "otel-v1", model: &encodeModel{otelV1: true}}, + } { + t.Run(mode.name, func(t *testing.T) { + resource, scope := testMetricResourceAndScope() + + metric := pmetric.NewMetric() + metric.SetName("system.cpu.usage") + dp := metric.SetEmptyGauge().DataPoints().AppendEmpty() + dp.SetDoubleValue(0.42) + dp.Attributes().PutStr("code.function", "doWork") + dp.Attributes().PutStr("code.function.name", "doWork") + + payload, err := mode.model.encodeMetric(resource, scope, "", metric, dp) + doc := encodeToMap(t, payload, err) + + attrs, ok := doc["attributes"].(map[string]any) + require.True(t, ok) + _, hasConcrete := attrs["code.function"] + assert.False(t, hasConcrete, "bare concrete code.function must be rewritten") + assert.Equal(t, "doWork", attrs["code.function.value"]) + assert.Equal(t, "doWork", attrs["code.function.name"]) + }) + } +} diff --git a/exporter/opensearchexporter/mock_test.go b/exporter/opensearchexporter/mock_test.go index 1a6fe57c49909..916de6c42468d 100644 --- a/exporter/opensearchexporter/mock_test.go +++ b/exporter/opensearchexporter/mock_test.go @@ -678,6 +678,54 @@ func TestOpenSearchOTelV1_ManageIndexTemplate(t *testing.T) { require.NoError(t, exporter.Shutdown(t.Context())) } +func TestOpenSearchSS4O_ManageIndexTemplate(t *testing.T) { + var templateRequests []string + var templateBodies []string + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead && strings.Contains(r.URL.Path, "_index_template") { + w.WriteHeader(http.StatusNotFound) + return + } + if r.Method == http.MethodPut && strings.Contains(r.URL.Path, "_index_template") { + templateRequests = append(templateRequests, r.URL.Path) + body, _ := io.ReadAll(r.Body) + templateBodies = append(templateBodies, string(body)) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"acknowledged": true}`)) + return + } + w.WriteHeader(http.StatusOK) + response, _ := os.ReadFile("testdata/opensearch-response-no-error.json") + _, _ = w.Write(response) + })) + defer ts.Close() + + cfg := withDefaultConfig(func(config *Config) { + config.ClientConfig.Endpoint = ts.URL + config.TimeoutSettings.Timeout = 0 + config.MappingsSettings.Mode = "ss4o" + config.MappingsSettings.ManageIndexTemplate = true + }) + + f := NewFactory() + exporter, err := f.CreateTraces(t.Context(), exportertest.NewNopSettings(metadata.Type), cfg) + require.NoError(t, err) + require.NoError(t, exporter.Start(t.Context(), componenttest.NewNopHost())) + + require.Len(t, templateRequests, 2) + assert.Contains(t, templateRequests[0], "ss4o-traces-index-template") + assert.Contains(t, templateRequests[1], "ss4o-logs-index-template") + // The installed templates must map attribute bags as flat_object so + // OpenSearch does not expand dotted attribute keys into nested objects. + assert.Contains(t, templateBodies[0], `"flat_object"`) + assert.Contains(t, templateBodies[0], "ss4o_traces-*") + assert.Contains(t, templateBodies[1], `"flat_object"`) + assert.Contains(t, templateBodies[1], "ss4o_logs-*") + + require.NoError(t, exporter.Shutdown(t.Context())) +} + func TestOpenSearchOTelV1_ManageIndexTemplate_Disabled(t *testing.T) { var templateRequests []string diff --git a/exporter/opensearchexporter/sso_log_exporter.go b/exporter/opensearchexporter/sso_log_exporter.go index 0909e7ce797d8..e304b33ffdf0f 100644 --- a/exporter/opensearchexporter/sso_log_exporter.go +++ b/exporter/opensearchexporter/sso_log_exporter.go @@ -81,7 +81,7 @@ func (l *logExporter) Start(ctx context.Context, host component.Host) error { if l.config.MappingsSettings.ManageIndexTemplate { tm := newTemplateManager(client, l.telemetry.Logger) - tm.ensureTemplates(ctx) + tm.ensureTemplates(ctx, l.config.MappingsSettings.Mode) } return nil diff --git a/exporter/opensearchexporter/sso_trace_exporter.go b/exporter/opensearchexporter/sso_trace_exporter.go index b07a818021061..d25e684e8b6c2 100644 --- a/exporter/opensearchexporter/sso_trace_exporter.go +++ b/exporter/opensearchexporter/sso_trace_exporter.go @@ -72,7 +72,7 @@ func (s *ssoTracesExporter) Start(ctx context.Context, host component.Host) erro if s.config.MappingsSettings.ManageIndexTemplate { tm := newTemplateManager(client, s.telemetry.Logger) - tm.ensureTemplates(ctx) + tm.ensureTemplates(ctx, s.config.MappingsSettings.Mode) } return nil