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

Add custom suffix to identifiers in filestream input when needed #26669

Merged
merged 9 commits into from
Jul 15, 2021
Merged

Add custom suffix to identifiers in filestream input when needed #26669

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a15739e
Add custom suffix to identifiers in filestream input
kvch Jun 28, 2021
d4182cb
pass suffix differently
kvch Jul 13, 2021
481b1af
address review notes
kvch Jul 13, 2021
17ba1a7
fix test code
kvch Jul 13, 2021
61ed885
minimize change
kvch Jul 13, 2021
016031b
more fix
kvch Jul 13, 2021
9947984
add error if multiple suffix is configured
kvch Jul 14, 2021
ba77a50
fix error :facepalm:
kvch Jul 14, 2021
d14cfb1
add tests for parser suffix
kvch Jul 15, 2021
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
2 changes: 1 addition & 1 deletion filebeat/input/filestream/environment_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,7 +281,7 @@ func (e *inputTestingEnvironment) getRegistryState(key string) (registryEntry, e
}

func getIDFromPath(filepath string, fi os.FileInfo) string {
identifier, _ := newINodeDeviceIdentifier(nil)
identifier, _ := newINodeDeviceIdentifier(nil, "")
src := identifier.GetSource(loginp.FSEvent{Info: fi, Op: loginp.OpCreate, NewPath: filepath})
return "filestream::.global::" + src.Name()
}
Expand Down
34 changes: 23 additions & 11 deletions filebeat/input/filestream/identifier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ var (
}
)

type identifierFactory func(*common.Config) (fileIdentifier, error)
type identifierFactory func(c *common.Config, suffix string) (fileIdentifier, error)

type fileIdentifier interface {
GetSource(loginp.FSEvent) fileSource
Expand All @@ -76,9 +76,9 @@ func (f fileSource) Name() string {
}

// newFileIdentifier creates a new state identifier for a log input.
func newFileIdentifier(ns *common.ConfigNamespace) (fileIdentifier, error) {
func newFileIdentifier(ns *common.ConfigNamespace, suffix string) (fileIdentifier, error) {
if ns == nil {
return newINodeDeviceIdentifier(nil)
return newINodeDeviceIdentifier(nil, suffix)
}

identifierType := ns.Name()
Expand All @@ -87,27 +87,33 @@ func newFileIdentifier(ns *common.ConfigNamespace) (fileIdentifier, error) {
return nil, fmt.Errorf("no such file_identity generator: %s", identifierType)
}

return f(ns.Config())
return f(ns.Config(), suffix)
}

type inodeDeviceIdentifier struct {
name string
name string
suffix string
kvch marked this conversation as resolved.
Show resolved Hide resolved
}

func newINodeDeviceIdentifier(_ *common.Config) (fileIdentifier, error) {
func newINodeDeviceIdentifier(_ *common.Config, suffix string) (fileIdentifier, error) {
return &inodeDeviceIdentifier{
name: nativeName,
name: nativeName,
suffix: suffix,
}, nil
}

func (i *inodeDeviceIdentifier) GetSource(e loginp.FSEvent) fileSource {
name := i.name + identitySep + file.GetOSState(e.Info).String()
if i.suffix != "" {
name += "-" + i.suffix
}
return fileSource{
info: e.Info,
newPath: e.NewPath,
oldPath: e.OldPath,
truncated: e.Op == loginp.OpTruncate,
archived: e.Op == loginp.OpArchived,
name: i.name + identitySep + file.GetOSState(e.Info).String(),
name: name,
identifierGenerator: i.name,
}
}
Expand All @@ -126,12 +132,14 @@ func (i *inodeDeviceIdentifier) Supports(f identifierFeature) bool {
}

type pathIdentifier struct {
name string
name string
suffix string
}

func newPathIdentifier(_ *common.Config) (fileIdentifier, error) {
func newPathIdentifier(_ *common.Config, suffix string) (fileIdentifier, error) {
return &pathIdentifier{
name: pathName,
name: pathName,
suffix: suffix,
}, nil
}

Expand All @@ -140,6 +148,10 @@ func (p *pathIdentifier) GetSource(e loginp.FSEvent) fileSource {
if e.Op == loginp.OpDelete {
path = e.OldPath
}
name := p.name + identitySep + path
if p.suffix != "" {
name += "-" + p.suffix
}
return fileSource{
info: e.Info,
newPath: e.NewPath,
Expand Down
11 changes: 8 additions & 3 deletions filebeat/input/filestream/identifier_inode_deviceid.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,14 @@ import (
type inodeMarkerIdentifier struct {
log *logp.Logger
name string
suffix string
markerPath string

markerFileLastModifitaion time.Time
markerTxt string
}

func newINodeMarkerIdentifier(cfg *common.Config) (fileIdentifier, error) {
func newINodeMarkerIdentifier(cfg *common.Config, suffix string) (fileIdentifier, error) {
var config struct {
MarkerPath string `config:"path" validate:"required"`
}
Expand All @@ -61,6 +62,7 @@ func newINodeMarkerIdentifier(cfg *common.Config) (fileIdentifier, error) {
return &inodeMarkerIdentifier{
log: logp.NewLogger("inode_marker_identifier_" + filepath.Base(config.MarkerPath)),
name: inodeMarkerName,
suffix: suffix,
markerPath: config.MarkerPath,
markerFileLastModifitaion: fi.ModTime(),
markerTxt: string(markerContent),
Expand Down Expand Up @@ -93,14 +95,17 @@ func (i *inodeMarkerIdentifier) markerContents() string {
}

func (i *inodeMarkerIdentifier) GetSource(e loginp.FSEvent) fileSource {
osstate := file.GetOSState(e.Info)
name := i.name + identitySep + file.GetOSState(e.Info).String() + "-" + i.markerContents()
if i.suffix != "" {
name += name + "-" + i.suffix
}
return fileSource{
info: e.Info,
newPath: e.NewPath,
oldPath: e.OldPath,
truncated: e.Op == loginp.OpTruncate,
archived: e.Op == loginp.OpArchived,
name: i.name + identitySep + osstate.InodeString() + "-" + i.markerContents(),
name: name,
identifierGenerator: i.name,
}
}
Expand Down
2 changes: 1 addition & 1 deletion filebeat/input/filestream/identifier_inode_deviceid_windows.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,6 @@ import (
"github.com/elastic/beats/v7/libbeat/common"
)

func newINodeMarkerIdentifier(cfg *common.Config) (fileIdentifier, error) {
func newINodeMarkerIdentifier(cfg *common.Config, suffix string) (fileIdentifier, error) {
return nil, fmt.Errorf("inode_deviceid is not supported on Windows")
}
28 changes: 26 additions & 2 deletions filebeat/input/filestream/identifier_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ type testFileIdentifierConfig struct {

func TestFileIdentifier(t *testing.T) {
t.Run("default file identifier", func(t *testing.T) {
identifier, err := newFileIdentifier(nil)
identifier, err := newFileIdentifier(nil, "")
require.NoError(t, err)
assert.Equal(t, DefaultIdentifierName, identifier.Name())

Expand All @@ -59,6 +59,30 @@ func TestFileIdentifier(t *testing.T) {
assert.Equal(t, identifier.Name()+"::"+file.GetOSState(fi).String(), src.Name())
})

t.Run("default file identifier with suffix", func(t *testing.T) {
identifier, err := newFileIdentifier(nil, "my-suffix")
require.NoError(t, err)
assert.Equal(t, DefaultIdentifierName, identifier.Name())

tmpFile, err := ioutil.TempFile("", "test_file_identifier_native")
if err != nil {
t.Fatalf("cannot create temporary file for test: %v", err)
}
defer os.Remove(tmpFile.Name())

fi, err := tmpFile.Stat()
if err != nil {
t.Fatalf("cannot stat temporary file for test: %v", err)
}

src := identifier.GetSource(loginp.FSEvent{
NewPath: tmpFile.Name(),
Info: fi,
})

assert.Equal(t, identifier.Name()+"::"+file.GetOSState(fi).String()+"-my-suffix", src.Name())
})

t.Run("path identifier", func(t *testing.T) {
c := common.MustNewConfigFrom(map[string]interface{}{
"identifier": map[string]interface{}{
Expand All @@ -69,7 +93,7 @@ func TestFileIdentifier(t *testing.T) {
err := c.Unpack(&cfg)
require.NoError(t, err)

identifier, err := newFileIdentifier(cfg.Identifier)
identifier, err := newFileIdentifier(cfg.Identifier, "")
require.NoError(t, err)
assert.Equal(t, pathName, identifier.Name())

Expand Down
6 changes: 5 additions & 1 deletion filebeat/input/filestream/prospector_creator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ func newProspector(config config) (loginp.Prospector, error) {
return nil, fmt.Errorf("error while creating filewatcher %v", err)
}

identifier, err := newFileIdentifier(config.FileIdentity)
identifier, err := newFileIdentifier(config.FileIdentity, getIdentifierSuffix(config))
if err != nil {
return nil, fmt.Errorf("error while creating file identifier: %v", err)
}
Expand Down Expand Up @@ -104,3 +104,7 @@ func newProspector(config config) (loginp.Prospector, error) {
}
return nil, fmt.Errorf("no such rotation method: %s", rotationMethod)
}

func getIdentifierSuffix(config config) string {
return config.Reader.Parsers.Suffix
}
2 changes: 1 addition & 1 deletion filebeat/input/filestream/prospector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -545,7 +545,7 @@ type renamedPathIdentifier struct {
func (p *renamedPathIdentifier) Supports(_ identifierFeature) bool { return true }

func mustPathIdentifier(renamed bool) fileIdentifier {
pathIdentifier, err := newPathIdentifier(nil)
pathIdentifier, err := newPathIdentifier(nil, "")
if err != nil {
panic(err)
}
Expand Down
7 changes: 7 additions & 0 deletions libbeat/reader/parser/parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ type CommonConfig struct {
}

type Config struct {
Suffix string
Copy link

Choose a reason for hiding this comment

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

maybe we want to name this field stream?

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 named it Suffix because from outside it is the string that is supposed to be added to the end of the state IDs. It does not matter if it contains a stream name or emojis describing the weather. The only thing that is relevant that it is a suffix.


pCfg CommonConfig
parsers []common.ConfigNamespace
}
Expand Down Expand Up @@ -79,6 +81,7 @@ func (c *Config) Unpack(cc *common.Config) error {
}

func NewConfig(pCfg CommonConfig, parsers []common.ConfigNamespace) (*Config, error) {
var suffix string
for _, ns := range parsers {
name := ns.Name()
switch name {
Expand All @@ -103,12 +106,16 @@ func NewConfig(pCfg CommonConfig, parsers []common.ConfigNamespace) (*Config, er
if err != nil {
return nil, fmt.Errorf("error while parsing container parser config: %+v", err)
}
if config.Stream != readjson.All {
suffix = config.Stream.String()
Copy link

Choose a reason for hiding this comment

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

What if suffix (the stream) is already set?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Only one suffix is supported at the moment.

Copy link

Choose a reason for hiding this comment

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

Maybe we should error then.

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 have added an error.

}
default:
return nil, fmt.Errorf("%s: %s", ErrNoSuchParser, name)
}
}

return &Config{
Suffix: suffix,
pCfg: pCfg,
parsers: parsers,
}, nil
Expand Down