Utilize resources (#105)

* Utilize resources

This commit makes a series of changes to make use of the new top-level
resource field.

The contents of `$resource` use the same resource standard as Open
Telemetry. See here for details: https://github.com/open-telemetry/opentelemetry-specification/tree/master/specification/resource/semantic_conventions

This adds support for accessing fields with dots in them using syntax like
`$record['field.with.dots']`.

This also modifies the `host_metadata`, `k8s_event_input`,
and `k8s_metadata_decorator` operators to add resource identifiers to
the entry.

Additionally, it adds support in the `google_cloud_output` for using the
resource field to add logs to a monitored resource. In the process, it
moves `google_cloud_output` into its own package for better organization
of its multiple files.



Includes a fix for a minor issue where rendering plugin ID in the template would set it as <no value> if using a default plugin ID.

Includes a fix for an issue where not all k8s events had a LastTimestamp, so now we prefer EventTimestamp, but fall back to LastTimestamp, then FirstTimestamp

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+### Added
+- Support for accessing the resource with fields
+- Support for using fields to select keys that contain dots like `$record['field.with.dots']`
+- `google_cloud_output` will use resource create a monitored resource for supported resource types (currently only k8s resources)
+### Changed
+- The operators `host_metadata`, `k8s_event_input`, and `k8s_metadata_decorator` will now use the top-level resource field
+
 ## [0.9.12] - 2020-08-25
 ### Changed
 - Agent is now embeddable with a default output

docs/types/field.md

@@ -3,7 +3,9 @@
 _Fields_ are the primary way to tell stanza which values of an entry to use in its operators.
 Most often, these will be things like fields to parse for a parser operator, or the field to write a new value to.
 
-Fields are `.`-delimited strings which allow you to select labels or records on the entry. Fields can currently be used to select labels or values on a record. To select a label, prefix your field with `$label.` such as with `$label.my_label`. For values on the record, use the prefix `$record.` such as `$record.my_value`.
+Fields are `.`-delimited strings which allow you to select labels or records on the entry. Fields can currently be used to select labels, values on a record, or resource values. To select a label, prefix your field with `$label` such as with `$label.my_label`. For values on the record, use the prefix `$record` such as `$record.my_value`. For resource values, use the prefix `$resource`.
+
+If a key contains a dot in it, a field can alternatively use bracket syntax for traversing through a map. For example, to select the key `k8s.cluster.name` on the entry's record, you can use the field `$record["k8s.cluster.name"]`.
 
 Record fields can be nested arbitrarily deeply, such as `$record.my_value.my_nested_value`.
 

entry/field.go

@@ -3,7 +3,6 @@ package entry
 import (
 	"encoding/json"
 	"fmt"
-	"strings"
 )
 
 // Field represents a potential field on an entry.
@@ -44,14 +43,22 @@ func (f *Field) UnmarshalYAML(unmarshal func(interface{}) error) error {
 }
 
 func fieldFromString(s string) (Field, error) {
-	split := strings.Split(s, ".")
+	split, err := splitField(s)
+	if err != nil {
+		return Field{}, fmt.Errorf("splitting field: %s", err)
+	}
 
 	switch split[0] {
 	case "$labels":
 		if len(split) != 2 {
 			return Field{}, fmt.Errorf("labels cannot be nested")
 		}
 		return Field{LabelField{split[1]}}, nil
+	case "$resource":
+		if len(split) != 2 {
+			return Field{}, fmt.Errorf("resource fields cannot be nested")
+		}
+		return Field{ResourceField{split[1]}}, nil
 	case "$record", "$":
 		return Field{RecordField{split[1:]}}, nil
 	default:
@@ -61,10 +68,90 @@ func fieldFromString(s string) (Field, error) {
 
 // MarshalJSON will marshal a field into JSON
 func (f Field) MarshalJSON() ([]byte, error) {
-	return []byte(fmt.Sprintf("\"%s\"", f.String())), nil
+	return []byte(fmt.Sprintf(`"%s"`, f.String())), nil
 }
 
 // MarshalYAML will marshal a field into YAML
 func (f Field) MarshalYAML() (interface{}, error) {
 	return f.String(), nil
 }
+
+type splitState uint
+
+const (
+	BEGIN splitState = iota
+	IN_BRACKET
+	IN_QUOTE
+	OUT_QUOTE
+	OUT_BRACKET
+	IN_UNBRACKETED_TOKEN
+)
+
+func splitField(s string) ([]string, error) {
+	fields := make([]string, 0, 1)
+
+	state := BEGIN
+	var quoteChar rune
+	var tokenStart int
+
+	for i, c := range s {
+		switch state {
+		case BEGIN:
+			if c == '[' {
+				state = IN_BRACKET
+				continue
+			}
+			tokenStart = i
+			state = IN_UNBRACKETED_TOKEN
+		case IN_BRACKET:
+			if !(c == '\'' || c == '"') {
+				return nil, fmt.Errorf("strings in brackets must be surrounded by quotes")
+			}
+			state = IN_QUOTE
+			quoteChar = c
+			tokenStart = i + 1
+		case IN_QUOTE:
+			if c == quoteChar {
+				fields = append(fields, s[tokenStart:i])
+				state = OUT_QUOTE
+			}
+		case OUT_QUOTE:
+			if c != ']' {
+				return nil, fmt.Errorf("found characters between closed quote and closing bracket")
+			}
+			state = OUT_BRACKET
+		case OUT_BRACKET:
+			if c == '.' {
+				state = IN_UNBRACKETED_TOKEN
+				tokenStart = i + 1
+			} else if c == '[' {
+				state = IN_BRACKET
+			} else {
+				return nil, fmt.Errorf("bracketed access must be followed by a dot or another bracketed access")
+			}
+		case IN_UNBRACKETED_TOKEN:
+			if c == '.' {
+				fields = append(fields, s[tokenStart:i])
+				tokenStart = i + 1
+			} else if c == '[' {
+				fields = append(fields, s[tokenStart:i])
+				state = IN_BRACKET
+			}
+		}
+	}
+
+	switch state {
+	case IN_BRACKET, OUT_QUOTE:
+		return nil, fmt.Errorf("found unclosed left bracket")
+	case IN_QUOTE:
+		if quoteChar == '"' {
+			return nil, fmt.Errorf("found unclosed double quote")
+		} else {
+			return nil, fmt.Errorf("found unclosed single quote")
+		}
+	case IN_UNBRACKETED_TOKEN:
+		fields = append(fields, s[tokenStart:])
+	}
+
+	return fields, nil
+}

entry/field_test.go

@@ -118,22 +118,57 @@ func TestFieldMarshalYAML(t *testing.T) {
 	cases := []struct {
 		name     string
 		input    interface{}
-		expected []byte
+		expected string
 	}{
 		{
 			"SimpleField",
 			NewRecordField("test1"),
-			[]byte("test1\n"),
+			"test1\n",
 		},
 		{
 			"ComplexField",
 			NewRecordField("test1", "test2"),
-			[]byte("test1.test2\n"),
+			"test1.test2\n",
 		},
 		{
 			"EmptyField",
 			NewRecordField(),
-			[]byte("$record\n"),
+			"$record\n",
+		},
+		{
+			"FieldWithDots",
+			NewRecordField("test.1"),
+			"$record['test.1']\n",
+		},
+		{
+			"FieldWithDotsThenNone",
+			NewRecordField("test.1", "test2"),
+			"$record['test.1']['test2']\n",
+		},
+		{
+			"FieldWithNoDotsThenDots",
+			NewRecordField("test1", "test.2"),
+			"$record['test1']['test.2']\n",
+		},
+		{
+			"LabelField",
+			NewLabelField("test1"),
+			"$labels.test1\n",
+		},
+		{
+			"LabelFieldWithDots",
+			NewLabelField("test.1"),
+			"$labels['test.1']\n",
+		},
+		{
+			"ResourceField",
+			NewResourceField("test1"),
+			"$resource.test1\n",
+		},
+		{
+			"ResourceFieldWithDots",
+			NewResourceField("test.1"),
+			"$resource['test.1']\n",
 		},
 	}
 
@@ -142,7 +177,48 @@ func TestFieldMarshalYAML(t *testing.T) {
 			res, err := yaml.Marshal(tc.input)
 			require.NoError(t, err)
 
-			require.Equal(t, tc.expected, res)
+			require.Equal(t, tc.expected, string(res))
+		})
+	}
+}
+
+func TestSplitField(t *testing.T) {
+	cases := []struct {
+		name      string
+		input     string
+		output    []string
+		expectErr bool
+	}{
+		{"Simple", "test", []string{"test"}, false},
+		{"Sub", "test.case", []string{"test", "case"}, false},
+		{"Root", "$", []string{"$"}, false},
+		{"RootWithSub", "$record.field", []string{"$record", "field"}, false},
+		{"RootWithTwoSub", "$record.field1.field2", []string{"$record", "field1", "field2"}, false},
+		{"BracketSyntaxSingleQuote", "['test']", []string{"test"}, false},
+		{"BracketSyntaxDoubleQuote", `["test"]`, []string{"test"}, false},
+		{"RootSubBracketSyntax", `$record["test"]`, []string{"$record", "test"}, false},
+		{"BracketThenDot", `$record["test1"].test2`, []string{"$record", "test1", "test2"}, false},
+		{"BracketThenBracket", `$record["test1"]["test2"]`, []string{"$record", "test1", "test2"}, false},
+		{"DotThenBracket", `$record.test1["test2"]`, []string{"$record", "test1", "test2"}, false},
+		{"DotsInBrackets", `$record["test1.test2"]`, []string{"$record", "test1.test2"}, false},
+		{"UnclosedBrackets", `$record["test1.test2"`, nil, true},
+		{"UnclosedQuotes", `$record["test1.test2]`, nil, true},
+		{"UnmatchedQuotes", `$record["test1.test2']`, nil, true},
+		{"BracketAtEnd", `$record[`, nil, true},
+		{"SingleQuoteAtEnd", `$record['`, nil, true},
+		{"DoubleQuoteAtEnd", `$record["`, nil, true},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := splitField(tc.input)
+			if tc.expectErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+
+			require.Equal(t, tc.output, s)
 		})
 	}
 }

entry/label_field.go

@@ -1,6 +1,9 @@
 package entry
 
-import "fmt"
+import (
+	"fmt"
+	"strings"
+)
 
 // LabelField is the path to an entry label
 type LabelField struct {
@@ -42,6 +45,9 @@ func (l LabelField) Delete(entry *Entry) (interface{}, bool) {
 }
 
 func (l LabelField) String() string {
+	if strings.Contains(l.key, ".") {
+		return fmt.Sprintf(`$labels['%s']`, l.key)
+	}
 	return "$labels." + l.key
 }
 

entry/record_field.go

@@ -210,7 +210,32 @@ func toJSONDot(field RecordField) string {
 		return "$record"
 	}
 
-	return strings.Join(field.Keys, ".")
+	containsDots := false
+	for _, key := range field.Keys {
+		if strings.Contains(key, ".") {
+			containsDots = true
+		}
+	}
+
+	var b strings.Builder
+	if containsDots {
+		b.WriteString("$record")
+		for _, key := range field.Keys {
+			b.WriteString(`['`)
+			b.WriteString(key)
+			b.WriteString(`']`)
+		}
+	} else {
+		for i, key := range field.Keys {
+			if i != 0 {
+				b.WriteString(".")
+			}
+			b.WriteString(key)
+		}
+
+	}
+
+	return b.String()
 }
 
 // NewRecordField creates a new field from an ordered array of keys.

entry/resource_field.go

@@ -0,0 +1,57 @@
+package entry
+
+import (
+	"fmt"
+	"strings"
+)
+
+// ResourceField is the path to an entry's resource key
+type ResourceField struct {
+	key string
+}
+
+// Get will return the resource value and a boolean indicating if it exists
+func (r ResourceField) Get(entry *Entry) (interface{}, bool) {
+	if entry.Resource == nil {
+		return "", false
+	}
+	val, ok := entry.Resource[r.key]
+	return val, ok
+}
+
+// Set will set the resource value on an entry
+func (r ResourceField) Set(entry *Entry, val interface{}) error {
+	if entry.Resource == nil {
+		entry.Resource = make(map[string]string, 1)
+	}
+
+	str, ok := val.(string)
+	if !ok {
+		return fmt.Errorf("cannot set a resource to a non-string value")
+	}
+	entry.Resource[r.key] = str
+	return nil
+}
+
+// Delete will delete a resource key from an entry
+func (r ResourceField) Delete(entry *Entry) (interface{}, bool) {
+	if entry.Resource == nil {
+		return "", false
+	}
+
+	val, ok := entry.Resource[r.key]
+	delete(entry.Resource, r.key)
+	return val, ok
+}
+
+func (r ResourceField) String() string {
+	if strings.Contains(r.key, ".") {
+		return fmt.Sprintf(`$resource['%s']`, r.key)
+	}
+	return "$resource." + r.key
+}
+
+// NewResourceField will creat a new resource field from a key
+func NewResourceField(key string) Field {
+	return Field{ResourceField{key}}
+}