-
Notifications
You must be signed in to change notification settings - Fork 141
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use metadata informer for ReplicaSets
- Loading branch information
Showing
3 changed files
with
207 additions
and
42 deletions.
There are no files selected for viewing
123 changes: 123 additions & 0 deletions
123
internal/pkg/composable/providers/kubernetes/metagen.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| package kubernetes | ||
|
|
||
| import ( | ||
| "github.com/elastic/elastic-agent-autodiscover/kubernetes" | ||
| "github.com/elastic/elastic-agent-autodiscover/kubernetes/metadata" | ||
| "github.com/elastic/elastic-agent-libs/config" | ||
| "github.com/elastic/elastic-agent-libs/mapstr" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| k8s "k8s.io/client-go/kubernetes" | ||
| "k8s.io/client-go/tools/cache" | ||
| ) | ||
|
|
||
| type ( | ||
| AddResourceMetadataConfig = metadata.AddResourceMetadataConfig | ||
| MetaGen = metadata.MetaGen | ||
| FieldOptions = metadata.FieldOptions | ||
| ) | ||
|
|
||
| // GetPodMetaGen is a wrapper function that creates a metaGen for pod resource and has embeeded | ||
| // nodeMetaGen and namespaceMetaGen | ||
| func GetPodMetaGen( | ||
| cfg *config.C, | ||
| podWatcher kubernetes.Watcher, | ||
| nodeWatcher kubernetes.Watcher, | ||
| namespaceWatcher kubernetes.Watcher, | ||
| replicasetWatcher kubernetes.Watcher, | ||
| jobWatcher kubernetes.Watcher, | ||
| metaConf *AddResourceMetadataConfig) MetaGen { | ||
|
|
||
| var nodeMetaGen, namespaceMetaGen, rsMetaGen, jobMetaGen MetaGen | ||
| if nodeWatcher != nil && metaConf.Node.Enabled() { | ||
| nodeMetaGen = metadata.NewNodeMetadataGenerator(metaConf.Node, nodeWatcher.Store(), nodeWatcher.Client()) | ||
| } | ||
| if namespaceWatcher != nil && metaConf.Namespace.Enabled() { | ||
| namespaceMetaGen = metadata.NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store(), namespaceWatcher.Client()) | ||
| } | ||
| if replicasetWatcher != nil && metaConf.Deployment { | ||
| // use our own implementation of this generator, which can avoid tracking the full ReplicaSet resource | ||
| // TODO: Remove this after upstreaming the change to the autodiscovery lib | ||
| rsMetaGen = NewReplicasetMetadataGenerator(cfg, replicasetWatcher.Store(), replicasetWatcher.Client()) | ||
| } | ||
| if jobWatcher != nil && metaConf.CronJob { | ||
| jobMetaGen = metadata.NewJobMetadataGenerator(cfg, jobWatcher.Store(), jobWatcher.Client()) | ||
| } | ||
| metaGen := metadata.NewPodMetadataGenerator( | ||
| cfg, | ||
| podWatcher.Store(), | ||
| podWatcher.Client(), | ||
| nodeMetaGen, | ||
| namespaceMetaGen, | ||
| rsMetaGen, | ||
| jobMetaGen, | ||
| metaConf) | ||
| return metaGen | ||
| } | ||
|
|
||
| const resourceType = "replicaset" | ||
|
|
||
| type replicaset struct { | ||
| store cache.Store | ||
| resource *metadata.Resource | ||
| } | ||
|
|
||
| // NewReplicasetMetadataGenerator creates a metagen for replicaset resources | ||
| func NewReplicasetMetadataGenerator(cfg *config.C, replicasets cache.Store, client k8s.Interface) MetaGen { | ||
| return &replicaset{ | ||
| resource: metadata.NewResourceMetadataGenerator(cfg, client), | ||
| store: replicasets, | ||
| } | ||
| } | ||
|
|
||
| // Generate generates replicaset metadata from a resource object | ||
| // Metadata map is in the following form: | ||
| // | ||
| // { | ||
| // "kubernetes": {}, | ||
| // "some.ecs.field": "asdf" | ||
| // } | ||
| // | ||
| // All Kubernetes fields that need to be stored under kuberentes. prefix are populetad by | ||
| // GenerateK8s method while fields that are part of ECS are generated by GenerateECS method | ||
| func (rs *replicaset) Generate(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M { | ||
| ecsFields := rs.GenerateECS(obj) | ||
| meta := mapstr.M{ | ||
| "kubernetes": rs.GenerateK8s(obj, opts...), | ||
| } | ||
| meta.DeepUpdate(ecsFields) | ||
| return meta | ||
| } | ||
|
|
||
| // GenerateECS generates replicaset ECS metadata from a resource object | ||
| func (rs *replicaset) GenerateECS(obj kubernetes.Resource) mapstr.M { | ||
| return rs.resource.GenerateECS(obj) | ||
| } | ||
|
|
||
| // GenerateK8s generates replicaset metadata from a resource object | ||
| func (rs *replicaset) GenerateK8s(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M { | ||
| _, ok := obj.(metav1.Object) // one of the changes from upstream autodiscovery | ||
| if !ok { | ||
| return nil | ||
| } | ||
|
|
||
| meta := rs.resource.GenerateK8s(resourceType, obj, opts...) | ||
| return meta | ||
| } | ||
|
|
||
| // GenerateFromName generates replicaset metadata from a replicaset name | ||
| func (rs *replicaset) GenerateFromName(name string, opts ...FieldOptions) mapstr.M { | ||
| if rs.store == nil { | ||
| return nil | ||
| } | ||
|
|
||
| if obj, ok, _ := rs.store.GetByKey(name); ok { | ||
| replicaSet, ok := obj.(kubernetes.Resource) // one of the changes from upstream autodiscovery | ||
| if !ok { | ||
| return nil | ||
| } | ||
|
|
||
| return rs.GenerateK8s(replicaSet, opts...) | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters