Skip to content

Commit

Permalink
Elasticsearch exporter: Add dedot support (#4579)
Browse files Browse the repository at this point in the history 
Add dedot support to Elasticsearch exporter. With all fields being sorted we iterate over the keys and dedot while serializing the document to JSON.

**Link to tracking Issue:** #1800 

**Testing:** Unit tests for dedot and flat serialization have been added.

**Documentation:** The dedot setting has already been documented. This PR adds the missing feature.
  • Loading branch information
Steffen Siering authored Aug 12, 2021
1 parent 957eb18 commit fcb5e97
Show file tree
Hide file tree
Showing 2 changed files with 220 additions and 7 deletions.
103 changes: 96 additions & 7 deletions exporter/elasticsearchexporter/internal/objmodel/objmodel.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,6 @@
package objmodel

import (
"errors"
"io"
"math"
"sort"
Expand Down Expand Up @@ -255,11 +254,7 @@ func (doc *Document) iterJSONFlat(w *json.Visitor) error {

for i := range doc.fields {
fld := &doc.fields[i]

// filter out empty values
if fld.value.kind == KindIgnore ||
fld.value.kind == KindNil ||
(fld.value.kind == KindArr && len(fld.value.arr) == 0) {
if fld.value.IsEmpty() {
continue
}

Expand All @@ -273,7 +268,74 @@ func (doc *Document) iterJSONFlat(w *json.Visitor) error {
}

func (doc *Document) iterJSONDedot(w *json.Visitor) error {
return errors.New("TODO")
objPrefix := ""
level := 0

w.OnObjectStart(-1, structform.AnyType)
defer w.OnObjectFinished()

for i := range doc.fields {
fld := &doc.fields[i]
if fld.value.IsEmpty() {
continue
}

key := fld.key
// decrease object level until last reported and current key have the same path prefix
for L := commonObjPrefix(key, objPrefix); L < len(objPrefix); {
for L > 0 && key[L-1] != '.' {
L--
}

// remove levels and append write list of outstanding '}' into the writer
if L > 0 {
for delta := objPrefix[L:]; len(delta) > 0; {
idx := strings.IndexByte(delta, '.')
if idx < 0 {
break
}

delta = delta[idx+1:]
level--
w.OnObjectFinished()
}

objPrefix = key[:L]
} else { // no common prefix, close all objects we reported so far.
for ; level > 0; level-- {
w.OnObjectFinished()
}
objPrefix = ""
}
}

// increase object level up to current field
for {
start := len(objPrefix)
idx := strings.IndexByte(key[start:], '.')
if idx < 0 {
break
}

level++
objPrefix = key[:len(objPrefix)+idx+1]
fieldName := key[start : start+idx]
w.OnKey(fieldName)
w.OnObjectStart(-1, structform.AnyType)
}

// report value
fieldName := key[len(objPrefix):]
w.OnKey(fieldName)
fld.value.iterJSON(w, true)
}

// close all pending object levels
for ; level > 0; level-- {
w.OnObjectFinished()
}

return nil
}

// StringValue create a new value from a string.
Expand Down Expand Up @@ -352,6 +414,19 @@ func (v *Value) Dedup() {
}
}

func (v *Value) IsEmpty() bool {
switch v.kind {
case KindNil, KindIgnore:
return true
case KindArr:
return len(v.arr) == 0
case KindObject:
return len(v.doc.fields) == 0
default:
return false
}
}

func (v *Value) iterJSON(w *json.Visitor, dedot bool) error {
switch v.kind {
case KindNil:
Expand Down Expand Up @@ -430,3 +505,17 @@ func flattenKey(path, key string) string {
}
return path + "." + key
}

func commonObjPrefix(a, b string) int {
end := len(a)
if alt := len(b); alt < end {
end = alt
}

for i := 0; i < end; i++ {
if a[i] != b[i] {
return i
}
}
return end
}
124 changes: 124 additions & 0 deletions exporter/elasticsearchexporter/internal/objmodel/objmodel_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,130 @@ func TestValue_FromAttribute(t *testing.T) {
}
}

func TestDocument_Serialize_Flat(t *testing.T) {
tests := map[string]struct {
doc Document
want string
}{
"no nesting with multiple fields": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a": pdata.NewAttributeValueString("test"),
"b": pdata.NewAttributeValueInt(1),
})),
want: `{"a":"test","b":1}`,
},
"shared prefix": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a.str": pdata.NewAttributeValueString("test"),
"a.i": pdata.NewAttributeValueInt(1),
})),
want: `{"a.i":1,"a.str":"test"}`,
},
"multiple namespaces with dot": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a.str": pdata.NewAttributeValueString("test"),
"b.i": pdata.NewAttributeValueInt(1),
})),
want: `{"a.str":"test","b.i":1}`,
},
"nested maps": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a": func() pdata.AttributeValue {
m := pdata.NewAttributeValueMap()
m.MapVal().InsertString("str", "test")
m.MapVal().InsertInt("i", 1)
return m
}(),
})),
want: `{"a.i":1,"a.str":"test"}`,
},
"multi-level nested namespace maps": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a": func() pdata.AttributeValue {
m := pdata.NewAttributeValueMap()
m.MapVal().InsertString("b.str", "test")
m.MapVal().InsertInt("i", 1)
return m
}(),
})),
want: `{"a.b.str":"test","a.i":1}`,
},
}

for name, test := range tests {
t.Run(name, func(t *testing.T) {
var buf strings.Builder
test.doc.Dedup()
err := test.doc.Serialize(&buf, false)
require.NoError(t, err)

assert.Equal(t, test.want, buf.String())
})
}
}

func TestDocument_Serialize_Dedot(t *testing.T) {
tests := map[string]struct {
doc Document
want string
}{
"no nesting with multiple fields": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a": pdata.NewAttributeValueString("test"),
"b": pdata.NewAttributeValueInt(1),
})),
want: `{"a":"test","b":1}`,
},
"shared prefix": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a.str": pdata.NewAttributeValueString("test"),
"a.i": pdata.NewAttributeValueInt(1),
})),
want: `{"a":{"i":1,"str":"test"}}`,
},
"multiple namespaces": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a.str": pdata.NewAttributeValueString("test"),
"b.i": pdata.NewAttributeValueInt(1),
})),
want: `{"a":{"str":"test"},"b":{"i":1}}`,
},
"nested maps": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a": func() pdata.AttributeValue {
m := pdata.NewAttributeValueMap()
m.MapVal().InsertString("str", "test")
m.MapVal().InsertInt("i", 1)
return m
}(),
})),
want: `{"a":{"i":1,"str":"test"}}`,
},
"multi-level nested namespace maps": {
doc: DocumentFromAttributes(pdata.NewAttributeMap().InitFromMap(map[string]pdata.AttributeValue{
"a": func() pdata.AttributeValue {
m := pdata.NewAttributeValueMap()
m.MapVal().InsertString("b.c.str", "test")
m.MapVal().InsertInt("i", 1)
return m
}(),
})),
want: `{"a":{"b":{"c":{"str":"test"}},"i":1}}`,
},
}

for name, test := range tests {
t.Run(name, func(t *testing.T) {
var buf strings.Builder
test.doc.Dedup()
err := test.doc.Serialize(&buf, true)
require.NoError(t, err)

assert.Equal(t, test.want, buf.String())
})
}
}

func TestValue_Serialize(t *testing.T) {
tests := map[string]struct {
value Value
Expand Down

0 comments on commit fcb5e97

Please sign in to comment.