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

Implement k8s secrets provider for Agent #24789

Merged
merged 25 commits into from
Apr 1, 2021
Merged

Implement k8s secrets provider for Agent #24789

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
996850e
wip
ChrsMark Mar 11, 2021
fa16391
save dynamic provider working state
ChrsMark Mar 26, 2021
0b582b3
Working state
ChrsMark Mar 26, 2021
a0ad168
fix config
ChrsMark Mar 26, 2021
bb4fc0c
fixes
ChrsMark Mar 26, 2021
a2d9051
Add code to fetch secret from k8s api
ChrsMark Mar 26, 2021
5e1dad2
fmt
ChrsMark Mar 29, 2021
be9f7b1
fix function call
ChrsMark Mar 29, 2021
898e9cb
clear code parts
ChrsMark Mar 29, 2021
c731628
improvements
ChrsMark Mar 29, 2021
5d20635
break when matched
ChrsMark Mar 29, 2021
da4a3d8
fix docstring
ChrsMark Mar 29, 2021
68d3ea7
fix broken tests
ChrsMark Mar 29, 2021
5db6cad
review changes
ChrsMark Mar 30, 2021
5a85afb
Add interface implementation guard
ChrsMark Mar 30, 2021
eea350a
Remove comment
ChrsMark Mar 30, 2021
f023688
wip
ChrsMark Mar 31, 2021
72cccc7
Add lookupNode function
ChrsMark Mar 31, 2021
ae81eb1
Fix package name
ChrsMark Apr 1, 2021
fb8a33e
Add changelog
ChrsMark Apr 1, 2021
2332aec
Add vars test
ChrsMark Apr 1, 2021
7a0bdcf
Merge remote-tracking branch 'upstream/master' into add_k8s_secrets
ChrsMark Apr 1, 2021
22b5857
Add test for k8s_secrets provider
ChrsMark Apr 1, 2021
7551ef5
Add extra test
ChrsMark Apr 1, 2021
e47bb5a
lint
ChrsMark Apr 1, 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
Empty file added x-pack/elastic-agent/fleet.yml.lock
View file
Open in desktop
Empty file.
2 changes: 1 addition & 1 deletion x-pack/elastic-agent/pkg/agent/application/pipeline/emitter/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ func NewController(
caps capabilities.Capability,
reloadables ...reloadable,
) *Controller {
init, _ := transpiler.NewVars(map[string]interface{}{})
init, _ := transpiler.NewVars(map[string]interface{}{}, nil)

return &Controller{
logger: log,
Expand Down
1 change: 1 addition & 0 deletions x-pack/elastic-agent/pkg/agent/cmd/include.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/env"
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/host"
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/kubernetes"
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/kubernetes_secrets"
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/local"
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/localdynamic"
_ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable/providers/path"
Expand Down
2 changes: 1 addition & 1 deletion x-pack/elastic-agent/pkg/agent/transpiler/ast_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1616,7 +1616,7 @@ func TestHash(t *testing.T) {
}

func mustMakeVars(mapping map[string]interface{}) *Vars {
v, err := NewVars(mapping)
v, err := NewVars(mapping, nil)
if err != nil {
panic(err)
}
Expand Down
2 changes: 1 addition & 1 deletion x-pack/elastic-agent/pkg/agent/transpiler/utils_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -737,7 +737,7 @@ func TestRenderInputs(t *testing.T) {
}

func mustMakeVarsP(mapping map[string]interface{}, processorKey string, processors Processors) *Vars {
v, err := NewVarsWithProcessors(mapping, processorKey, processors)
v, err := NewVarsWithProcessors(mapping, processorKey, processors, nil)
if err != nil {
panic(err)
}
Expand Down
36 changes: 28 additions & 8 deletions x-pack/elastic-agent/pkg/agent/transpiler/vars.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ import (
"regexp"
"strings"
"unicode"

"github.com/elastic/beats/v7/libbeat/common"

"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/composable"
)

var varsRegex = regexp.MustCompile(`\${([\p{L}\d\s\\\-_|.'"]*)}`)
Expand All @@ -18,23 +22,24 @@ var ErrNoMatch = fmt.Errorf("no matching vars")

// Vars is a context of variables that also contain a list of processors that go with the mapping.
type Vars struct {
tree *AST
processorsKey string
processors Processors
tree *AST
processorsKey string
processors Processors
fetchContextProviders common.MapStr
}

// NewVars returns a new instance of vars.
func NewVars(mapping map[string]interface{}) (*Vars, error) {
return NewVarsWithProcessors(mapping, "", nil)
func NewVars(mapping map[string]interface{}, fetchContextProviders common.MapStr) (*Vars, error) {
return NewVarsWithProcessors(mapping, "", nil, fetchContextProviders)
}

// NewVarsWithProcessors returns a new instance of vars with attachment of processors.
func NewVarsWithProcessors(mapping map[string]interface{}, processorKey string, processors Processors) (*Vars, error) {
func NewVarsWithProcessors(mapping map[string]interface{}, processorKey string, processors Processors, fetchContextProviders common.MapStr) (*Vars, error) {
tree, err := NewAST(mapping)
if err != nil {
return nil, err
}
return &Vars{tree, processorKey, processors}, nil
return &Vars{tree, processorKey, processors, fetchContextProviders}, nil
}

// Replace returns a new value based on variable replacement.
Expand All @@ -44,7 +49,6 @@ func (v *Vars) Replace(value string) (Node, error) {
if !validBrackets(value, matchIdxs) {
return nil, fmt.Errorf("starting ${ is missing ending }")
}

result := ""
lastIndex := 0
for _, r := range matchIdxs {
Expand All @@ -55,6 +59,22 @@ func (v *Vars) Replace(value string) (Node, error) {
}
set := false
for _, val := range vars {
for providerName, provider := range v.fetchContextProviders {
if varPrefixMatched(val.Value(), providerName) {
fetchProvider := provider.(composable.FetchContextProvider)
fval, found := fetchProvider.Fetch(val.Value())
if found {
result += value[lastIndex:r[0]] + fval
set = true
break
} else {
return NewStrVal(""), ErrNoMatch
}
}
}
if set {
continue
}
Copy link
Contributor

Choose a reason for hiding this comment

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

Is it possible that getting a value from a fetch context provider returns something other than a string?

Below you can see that it will replace full objects if its a varString. I worry based on placing this before the switch val.(type) we are missing that as a possibility with a fetch context provider.

Copy link
Member Author

Choose a reason for hiding this comment

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

I see, however no strong opinion here since I'm not super familiar with the combinations that could occur. Do you think that removing continue and being able to get to the switch block after the Fetch() action would be sth that we want?

Copy link
Contributor

Choose a reason for hiding this comment

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

Now that I took a closer look at this file. This is definitely in the wrong place. At its current location it would try to find a constant string in a fetch context provider, we do not want that.

Looking at it, this needs to be removed from here and moved into the:

func (v *Vars) Lookup(name string) (interface{}, bool) {
	return v.tree.Lookup(name)
}

Leaving this function untouched.

Copy link
Member Author

@ChrsMark ChrsMark Mar 31, 2021 •
edited
Loading

Choose a reason for hiding this comment

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

@blakerouse I tried to move (see f023688) the code inside func (v *Vars) Lookup(name string) (interface{}, bool) method of the same file but it does not seem to work (using same configuration so far) and I'm not sure that I can follow the flow here. I see that the method is called at

beats/x-pack/elastic-agent/pkg/eql/visitor.go

Line 281 in 01eb297

val, ok := v.vars.Lookup(name)
and when debugging the flow with additional debug messages I cannot find anything useful so far. Could you please provide more content here?

Copy link
Contributor

Choose a reason for hiding this comment

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

I think I see the issue. You need to do the following:

Add the following function to Vars:

// lookupNode performs a lookup on the AST, but keeps the result as a `Node`.
//
// This is different from `Lookup` which returns the actual type, not the AST type.
func (v *Vars) lookupNode(name string) (Node, bool) {
	// check if the value can be retrieved from a FetchContextProvider
	for providerName, provider := range v.fetchContextProviders {
		if varPrefixMatched(name, providerName) {
			fetchProvider := provider.(composable.FetchContextProvider)
			fval, found := fetchProvider.Fetch(name)
			if found {
				return &StrVal{value: fval}, true
			} else {
				return &StrVal{value: ""}, false
			}
		}
	}
	// lookup in the AST tree
	return Lookup(v.tree, name)
}

Then change node, ok := Lookup(v.tree, val.Value()) to node, ok := v.lookupNode(val.Value()).

See if that works.

Copy link
Member Author

Choose a reason for hiding this comment

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

👍🏼 This fixes the issue, thank you!

switch val.(type) {
case *constString:
result += value[lastIndex:r[0]] + val.Value()
Expand Down
3 changes: 2 additions & 1 deletion x-pack/elastic-agent/pkg/agent/transpiler/vars_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,8 @@ func TestVars_ReplaceWithProcessors(t *testing.T) {
},
},
"dynamic",
processers)
processers,
nil)
ChrsMark marked this conversation as resolved.
Show resolved Hide resolved
require.NoError(t, err)

res, err := vars.Replace("${testing.key1}")
Expand Down
18 changes: 2 additions & 16 deletions x-pack/elastic-agent/pkg/composable/context.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,30 +5,16 @@
package composable

import (
"context"
"fmt"
"strings"

"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/config"
corecomp "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
)

// ContextProviderComm is the interface that a context provider uses to communicate back to Elastic Agent.
type ContextProviderComm interface {
context.Context

// Set sets the current mapping for this context.
Set(map[string]interface{}) error
}

// ContextProvider is the interface that a context provider must implement.
type ContextProvider interface {
// Run runs the context provider.
Run(ContextProviderComm) error
}

// ContextProviderBuilder creates a new context provider based on the given config and returns it.
type ContextProviderBuilder func(log *logger.Logger, config *config.Config) (ContextProvider, error)
type ContextProviderBuilder func(log *logger.Logger, config *config.Config) (corecomp.ContextProvider, error)

// AddContextProvider adds a new ContextProviderBuilder
func (r *providerRegistry) AddContextProvider(name string, builder ContextProviderBuilder) error {
Expand Down
22 changes: 15 additions & 7 deletions x-pack/elastic-agent/pkg/composable/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,16 +8,18 @@ import (
"context"
"encoding/json"
"fmt"
"strings"

"reflect"
"sort"
"strings"
"sync"
"time"

"github.com/elastic/beats/v7/libbeat/common"

"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/transpiler"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/config"
corecomp "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
)

Expand Down Expand Up @@ -98,6 +100,9 @@ func (c *controller) Run(ctx context.Context, cb VarsCallback) error {
notify := make(chan bool, 5000)
localCtx, cancel := context.WithCancel(ctx)

//fetchContextProviders := make(map[string]corecomp.ContextProvider, len(FetchContextProviders))
fetchContextProviders := common.MapStr{}

// run all the enabled context providers
for name, state := range c.contextProviders {
state.Context = localCtx
Expand All @@ -107,6 +112,9 @@ func (c *controller) Run(ctx context.Context, cb VarsCallback) error {
cancel()
return errors.New(err, fmt.Sprintf("failed to run provider '%s'", name), errors.TypeConfig, errors.M("provider", name))
}
if p, ok := state.provider.(corecomp.FetchContextProvider); ok {
fetchContextProviders.Put(name, p)
}
ChrsMark marked this conversation as resolved.
Show resolved Hide resolved
}

// run all the enabled dynamic providers
Expand Down Expand Up @@ -151,15 +159,15 @@ func (c *controller) Run(ctx context.Context, cb VarsCallback) error {
mapping[name] = state.Current()
}
// this is ensured not to error, by how the mappings states are verified
vars[0], _ = transpiler.NewVars(mapping)
vars[0], _ = transpiler.NewVars(mapping, fetchContextProviders)

// add to the vars list for each dynamic providers mappings
for name, state := range c.dynamicProviders {
for _, mappings := range state.Mappings() {
local, _ := cloneMap(mapping) // will not fail; already been successfully cloned once
local[name] = mappings.mapping
// this is ensured not to error, by how the mappings states are verified
v, _ := transpiler.NewVarsWithProcessors(local, name, mappings.processors)
v, _ := transpiler.NewVarsWithProcessors(local, name, mappings.processors, fetchContextProviders)
vars = append(vars, v)
}
}
Expand All @@ -175,7 +183,7 @@ func (c *controller) Run(ctx context.Context, cb VarsCallback) error {
type contextProviderState struct {
context.Context

provider ContextProvider
provider corecomp.ContextProvider
lock sync.RWMutex
mapping map[string]interface{}
signal chan bool
Expand All @@ -189,7 +197,7 @@ func (c *contextProviderState) Set(mapping map[string]interface{}) error {
return err
}
// ensure creating vars will not error
_, err = transpiler.NewVars(mapping)
_, err = transpiler.NewVars(mapping, nil)
if err != nil {
return err
}
Expand Down Expand Up @@ -244,7 +252,7 @@ func (c *dynamicProviderState) AddOrUpdate(id string, priority int, mapping map[
return err
}
// ensure creating vars will not error
_, err = transpiler.NewVars(mapping)
_, err = transpiler.NewVars(mapping, nil)
if err != nil {
return err
}
Expand Down
5 changes: 3 additions & 2 deletions x-pack/elastic-agent/pkg/composable/providers/agent/agent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/config"
corecomp "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/composable"
ChrsMark marked this conversation as resolved.
Show resolved Hide resolved
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/release"
)
Expand All @@ -20,7 +21,7 @@ func init() {
type contextProvider struct{}

// Run runs the Agent context provider.
func (*contextProvider) Run(comm composable.ContextProviderComm) error {
func (*contextProvider) Run(comm corecomp.ContextProviderComm) error {
a, err := info.NewAgentInfo()
if err != nil {
return err
Expand All @@ -41,6 +42,6 @@ func (*contextProvider) Run(comm composable.ContextProviderComm) error {
}

// ContextProviderBuilder builds the context provider.
func ContextProviderBuilder(_ *logger.Logger, _ *config.Config) (composable.ContextProvider, error) {
func ContextProviderBuilder(_ *logger.Logger, _ *config.Config) (corecomp.ContextProvider, error) {
return &contextProvider{}, nil
}
5 changes: 3 additions & 2 deletions x-pack/elastic-agent/pkg/composable/providers/env/env.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/config"
corecomp "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
)

Expand All @@ -21,7 +22,7 @@ func init() {
type contextProvider struct{}

// Run runs the environment context provider.
func (*contextProvider) Run(comm composable.ContextProviderComm) error {
func (*contextProvider) Run(comm corecomp.ContextProviderComm) error {
err := comm.Set(getEnvMapping())
if err != nil {
return errors.New(err, "failed to set mapping", errors.TypeUnexpected)
Expand All @@ -30,7 +31,7 @@ func (*contextProvider) Run(comm composable.ContextProviderComm) error {
}

// ContextProviderBuilder builds the context provider.
func ContextProviderBuilder(_ *logger.Logger, _ *config.Config) (composable.ContextProvider, error) {
func ContextProviderBuilder(_ *logger.Logger, _ *config.Config) (corecomp.ContextProvider, error) {
return &contextProvider{}, nil
}

Expand Down
5 changes: 3 additions & 2 deletions x-pack/elastic-agent/pkg/composable/providers/host/host.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/config"
corecomp "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/composable"
"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
)

Expand All @@ -38,7 +39,7 @@ type contextProvider struct {
}

// Run runs the environment context provider.
func (c *contextProvider) Run(comm composable.ContextProviderComm) error {
func (c *contextProvider) Run(comm corecomp.ContextProviderComm) error {
current, err := c.fetcher()
if err != nil {
return err
Expand Down Expand Up @@ -79,7 +80,7 @@ func (c *contextProvider) Run(comm composable.ContextProviderComm) error {
}

// ContextProviderBuilder builds the context provider.
func ContextProviderBuilder(log *logger.Logger, c *config.Config) (composable.ContextProvider, error) {
func ContextProviderBuilder(log *logger.Logger, c *config.Config) (corecomp.ContextProvider, error) {
p := &contextProvider{
logger: log,
fetcher: getHostInfo,
Expand Down
13 changes: 13 additions & 0 deletions x-pack/elastic-agent/pkg/composable/providers/kubernetes_secrets/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
// or more contributor license agreements. Licensed under the Elastic License;
// you may not use this file except in compliance with the Elastic License.

// TODO review the need for this
// +build linux darwin windows

package kubernetes_secrets

// Config for kubernetes provider
type Config struct {
KubeConfig string `config:"kube_config"`
}
Loading