file_input add mapstructure support (#25)

* file_input add mapstructure support

Co-authored-by: Daniel Jaglowski <[email protected]>

entry/field.go

@@ -179,5 +179,9 @@ func splitField(s string) ([]string, error) {
 		fields = append(fields, s[tokenStart:])
 	}
 
+	if len(fields) == 0 {
+		return nil, fmt.Errorf("fields size is 0")
+	}
+
 	return fields, nil
 }

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.0.2
 	github.com/json-iterator/go v1.1.10
 	github.com/kr/text v0.2.0 // indirect
+	github.com/mitchellh/mapstructure v1.4.1
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect

go.sum

@@ -36,6 +36,8 @@ github.com/lucasb-eyer/go-colorful v1.0.2/go.mod h1:0MS4r+7BZKSJ5mw4/S5MPN+qHFF1
 github.com/lucasb-eyer/go-colorful v1.0.3/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
 github.com/mattn/go-runewidth v0.0.4/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
 github.com/mattn/go-runewidth v0.0.8/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
+github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag=
+github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=

operator/builtin/input/file/config.go

@@ -55,26 +55,26 @@ func NewInputConfig(operatorID string) *InputConfig {
 
 // InputConfig is the configuration of a file input operator
 type InputConfig struct {
-	helper.InputConfig `yaml:",inline"`
-
-	Include []string `json:"include,omitempty" yaml:"include,omitempty"`
-	Exclude []string `json:"exclude,omitempty" yaml:"exclude,omitempty"`
-
-	PollInterval       helper.Duration  `json:"poll_interval,omitempty"        yaml:"poll_interval,omitempty"`
-	Multiline          *MultilineConfig `json:"multiline,omitempty"            yaml:"multiline,omitempty"`
-	IncludeFileName    bool             `json:"include_file_name,omitempty"    yaml:"include_file_name,omitempty"`
-	IncludeFilePath    bool             `json:"include_file_path,omitempty"    yaml:"include_file_path,omitempty"`
-	StartAt            string           `json:"start_at,omitempty"             yaml:"start_at,omitempty"`
-	FingerprintSize    helper.ByteSize  `json:"fingerprint_size,omitempty"     yaml:"fingerprint_size,omitempty"`
-	MaxLogSize         helper.ByteSize  `json:"max_log_size,omitempty"         yaml:"max_log_size,omitempty"`
-	MaxConcurrentFiles int              `json:"max_concurrent_files,omitempty" yaml:"max_concurrent_files,omitempty"`
-	Encoding           string           `json:"encoding,omitempty"             yaml:"encoding,omitempty"`
+	helper.InputConfig `mapstructure:",squash" yaml:",inline"`
+
+	Include []string `mapstructure:"include,omitempty" json:"include,omitempty" yaml:"include,omitempty"`
+	Exclude []string `mapstructure:"exclude,omitempty" json:"exclude,omitempty" yaml:"exclude,omitempty"`
+
+	PollInterval       helper.Duration  `mapstructure:"poll_interval,omitempty"         json:"poll_interval,omitempty"        yaml:"poll_interval,omitempty"`
+	Multiline          *MultilineConfig `mapstructure:"multiline,omitempty"             json:"multiline,omitempty"            yaml:"multiline,omitempty"`
+	IncludeFileName    bool             `mapstructure:"include_file_name,omitempty"     json:"include_file_name,omitempty"    yaml:"include_file_name,omitempty"`
+	IncludeFilePath    bool             `mapstructure:"include_file_path,omitempty"     json:"include_file_path,omitempty"    yaml:"include_file_path,omitempty"`
+	StartAt            string           `mapstructure:"start_at,omitempty"              json:"start_at,omitempty"             yaml:"start_at,omitempty"`
+	FingerprintSize    helper.ByteSize  `mapstructure:"fingerprint_size,omitempty"      json:"fingerprint_size,omitempty"     yaml:"fingerprint_size,omitempty"`
+	MaxLogSize         helper.ByteSize  `mapstructure:"max_log_size,omitempty"          json:"max_log_size,omitempty"         yaml:"max_log_size,omitempty"`
+	MaxConcurrentFiles int              `mapstructure:"max_concurrent_files,omitempty"  json:"max_concurrent_files,omitempty" yaml:"max_concurrent_files,omitempty"`
+	Encoding           string           `mapstructure:"encoding,omitempty"              json:"encoding,omitempty"             yaml:"encoding,omitempty"`
 }
 
 // MultilineConfig is the configuration a multiline operation
 type MultilineConfig struct {
-	LineStartPattern string `json:"line_start_pattern" yaml:"line_start_pattern"`
-	LineEndPattern   string `json:"line_end_pattern"   yaml:"line_end_pattern"`
+	LineStartPattern string `mapstructure:"line_start_pattern"  json:"line_start_pattern" yaml:"line_start_pattern"`
+	LineEndPattern   string `mapstructure:"line_end_pattern"    json:"line_end_pattern"   yaml:"line_end_pattern"`
 }
 
 // Build will build a file input operator from the supplied configuration

operator/builtin/input/file/config_test.go

@@ -21,9 +21,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper"
+	"github.com/mitchellh/mapstructure"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v2"
+
+	"github.com/open-telemetry/opentelemetry-log-collection/entry"
+	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper"
 )
 
 type testCase struct {
@@ -205,18 +208,17 @@ func TestConfig(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			cfgFromYaml, err := configFromFileViaYaml(t, path.Join(".", "testdata", fmt.Sprintf("%s.yaml", tc.name)))
+			cfgFromYaml, err := configFromFileViaYaml(path.Join(".", "testdata", fmt.Sprintf("%s.yaml", tc.name)))
 			require.NoError(t, err)
 			require.Equal(t, tc.expect, cfgFromYaml)
-
-			// TODO cfgFromMapstructure, err := configFromFileViaYaml(t, path.Join(".", "testdata", fmt.Sprintf("%s.yaml", tc.name)))
-			// TODO require.NoError(t, err)
-			// TODO require.Equal(t, tc.expect, cfgFromMapstructure)
+			cfgFromMapstructure, err := configFromFileViaMapstructure(path.Join(".", "testdata", fmt.Sprintf("%s.yaml", tc.name)))
+			require.NoError(t, err)
+			require.Equal(t, tc.expect, cfgFromMapstructure)
 		})
 	}
 }
 
-func configFromFileViaYaml(t *testing.T, file string) (*InputConfig, error) {
+func configFromFileViaYaml(file string) (*InputConfig, error) {
 	bytes, err := ioutil.ReadFile(file)
 	if err != nil {
 		return nil, fmt.Errorf("could not find config file: %s", err)
@@ -230,8 +232,115 @@ func configFromFileViaYaml(t *testing.T, file string) (*InputConfig, error) {
 	return config, nil
 }
 
-// TODO func configFromFileViaMapstructure(t *testing.T, file string) (*InputConfig, error) {}
+func configFromFileViaMapstructure(file string) (*InputConfig, error) {
+	bytes, err := ioutil.ReadFile(file)
+	if err != nil {
+		return nil, fmt.Errorf("could not find config file: %s", err)
+	}
+
+	raw := map[string]interface{}{}
+
+	if err := yaml.Unmarshal(bytes, raw); err != nil {
+		return nil, fmt.Errorf("failed to read data from yaml: %s", err)
+	}
+
+	cfg := defaultCfg()
+	dc := &mapstructure.DecoderConfig{Result: cfg, DecodeHook: helper.JSONUnmarshalerHook()}
+	ms, err := mapstructure.NewDecoder(dc)
+	if err != nil {
+		return nil, err
+	}
+	err = ms.Decode(raw)
+	if err != nil {
+		return nil, err
+	}
+	return cfg, nil
+
+}
 
 func defaultCfg() *InputConfig {
 	return NewInputConfig("file_input")
 }
+
+func NewTestInputConfig() *InputConfig {
+	cfg := NewInputConfig("config_test")
+	cfg.WriteTo = entry.NewRecordField([]string{}...)
+	cfg.Include = []string{"i1", "i2"}
+	cfg.Exclude = []string{"e1", "e2"}
+	cfg.Multiline = &MultilineConfig{"start", "end"}
+	cfg.FingerprintSize = 1024
+	cfg.Encoding = "utf16"
+	return cfg
+}
+
+func TestMapStructureDecodeConfigWithHook(t *testing.T) {
+	expect := NewTestInputConfig()
+	input := map[string]interface{}{
+		// InputConfig
+		"id":       "config_test",
+		"type":     "file_input",
+		"write_to": "$",
+		"labels": map[string]interface{}{
+		},
+		"resource": map[string]interface{}{
+		},
+
+		"include":       expect.Include,
+		"exclude":       expect.Exclude,
+		"poll_interval": 0.2,
+		"multiline": map[string]interface{}{
+			"line_start_pattern": expect.Multiline.LineStartPattern,
+			"line_end_pattern":   expect.Multiline.LineEndPattern,
+		},
+		"include_file_name":    true,
+		"include_file_path":    false,
+		"start_at":             "end",
+		"fingerprint_size":     "1024",
+		"max_log_size":         "1mib",
+		"max_concurrent_files": 1024,
+		"encoding":             "utf16",
+	}
+
+	var actual InputConfig
+	dc := &mapstructure.DecoderConfig{Result: &actual, DecodeHook: helper.JSONUnmarshalerHook()}
+	ms, err := mapstructure.NewDecoder(dc)
+	require.NoError(t, err)
+	err = ms.Decode(input)
+	require.NoError(t, err)
+	require.Equal(t, expect, &actual)
+}
+
+func TestMapStructureDecodeConfig(t *testing.T) {
+	expect := NewTestInputConfig()
+	input := map[string]interface{}{
+		// InputConfig
+		"id":       "config_test",
+		"type":     "file_input",
+		"write_to": entry.NewRecordField([]string{}...),
+		"labels": map[string]interface{}{
+		},
+		"resource": map[string]interface{}{
+		},
+		"include": expect.Include,
+		"exclude": expect.Exclude,
+		"poll_interval": map[string]interface{}{
+			"Duration": 200 * 1000 * 1000,
+		},
+		"multiline": map[string]interface{}{
+			"line_start_pattern": expect.Multiline.LineStartPattern,
+			"line_end_pattern":   expect.Multiline.LineEndPattern,
+		},
+		"include_file_name":    true,
+		"include_file_path":    false,
+		"start_at":             "end",
+		"fingerprint_size":     1024,
+		"max_log_size":         1024 * 1024,
+		"max_concurrent_files": 1024,
+		"encoding":             "utf16",
+	}
+
+	var actual InputConfig
+	err := mapstructure.Decode(input, &actual)
+	require.NoError(t, err)
+	require.Equal(t, expect, &actual)
+}

operator/helper/identifier.go

@@ -27,7 +27,7 @@ func NewIdentifierConfig() IdentifierConfig {
 
 // IdentifierConfig is the configuration of a resource identifier
 type IdentifierConfig struct {
-	Resource map[string]ExprStringConfig `json:"resource" yaml:"resource"`
+	Resource map[string]ExprStringConfig `mapstructure:"resource" json:"resource" yaml:"resource"`
 }
 
 // Build will build an identifier from the supplied configuration

operator/helper/input.go

@@ -35,10 +35,10 @@ func NewInputConfig(operatorID, operatorType string) InputConfig {
 
 // InputConfig provides a basic implementation of an input operator config.
 type InputConfig struct {
-	LabelerConfig    `yaml:",inline"`
-	IdentifierConfig `yaml:",inline"`
-	WriterConfig     `yaml:",inline"`
-	WriteTo          entry.Field `json:"write_to" yaml:"write_to"`
+	LabelerConfig    `mapstructure:",squash" yaml:",inline"`
+	IdentifierConfig `mapstructure:",squash" yaml:",inline"`
+	WriterConfig     `mapstructure:",squash" yaml:",inline"`
+	WriteTo          entry.Field `mapstructure:"write_to" json:"write_to" yaml:"write_to"`
 }
 
 // Build will build a base producer.

operator/helper/labeler.go

@@ -27,7 +27,7 @@ func NewLabelerConfig() LabelerConfig {
 
 // LabelerConfig is the configuration of a labeler
 type LabelerConfig struct {
-	Labels map[string]ExprStringConfig `json:"labels" yaml:"labels"`
+	Labels map[string]ExprStringConfig `mapstructure:"labels" json:"labels" yaml:"labels"`
 }
 
 // Build will build a labeler from the supplied configuration

operator/helper/mapstructure.go

@@ -0,0 +1,52 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package helper
+
+import (
+	"encoding/json"
+	"reflect"
+
+	"github.com/mitchellh/mapstructure"
+)
+
+// make mapstructure use struct UnmarshalJSON to decode
+func JSONUnmarshalerHook() mapstructure.DecodeHookFunc {
+	return func(from reflect.Value, to reflect.Value) (interface{}, error) {
+		if to.CanAddr() {
+			to = to.Addr()
+		}
+
+		// If the destination implements the unmarshaling interface
+		u, ok := to.Interface().(json.Unmarshaler)
+		if !ok {
+			return from.Interface(), nil
+		}
+
+		// If it is nil and a pointer, create and assign the target value first
+		if to.IsNil() && to.Type().Kind() == reflect.Ptr {
+			to.Set(reflect.New(to.Type().Elem()))
+			u = to.Interface().(json.Unmarshaler)
+		}
+		v := from.Interface()
+		bytes, err := json.Marshal(v)
+		if err != nil {
+			return nil, err
+		}
+		if err := u.UnmarshalJSON(bytes); err != nil {
+			return to.Interface(), err
+		}
+		return to.Interface(), nil
+	}
+}

operator/helper/operator.go

@@ -30,8 +30,8 @@ func NewBasicConfig(operatorID, operatorType string) BasicConfig {
 
 // BasicConfig provides a basic implemention for an operator config.
 type BasicConfig struct {
-	OperatorID   string `json:"id"   yaml:"id"`
-	OperatorType string `json:"type" yaml:"type"`
+	OperatorID   string `mapstructure:"id"   json:"id"   yaml:"id"`
+	OperatorType string `mapstructure:"type" json:"type" yaml:"type"`
 }
 
 // ID will return the operator id.

operator/helper/writer.go

@@ -32,8 +32,8 @@ func NewWriterConfig(operatorID, operatorType string) WriterConfig {
 
 // WriterConfig is the configuration of a writer operator.
 type WriterConfig struct {
-	BasicConfig `yaml:",inline"`
-	OutputIDs   OutputIDs `json:"output" yaml:"output"`
+	BasicConfig `mapstructure:",squash" yaml:",inline"`
+	OutputIDs   OutputIDs `mapstructure:"output" json:"output" yaml:"output"`
 }
 
 // Build will build a writer operator from the config.