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

Use replicaset and Job MetaGen based on watchers #35483

Merged
merged 17 commits into from
May 30, 2023
Merged

Use replicaset and Job MetaGen based on watchers #35483

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
6cabaeb
Use rs MetaGen based on watchers
ChrsMark May 16, 2023
05d8ab1
Use job MetaGen based on watchers
ChrsMark May 16, 2023
e86b851
Start watchers properly
ChrsMark May 17, 2023
13ddadc
Beautify code
ChrsMark May 17, 2023
d4e5e0c
Start/Stop replicaset and job watchers properly
ChrsMark May 17, 2023
0405168
Only create watchers if needed
ChrsMark May 18, 2023
739491c
Upgrade autodiscovery library version
ChrsMark May 26, 2023
43eab99
Merge remote-tracking branch 'upstream/main' into rs_watchers
ChrsMark May 26, 2023
b2b56bb
Upgrade autodiscovery library version
ChrsMark May 26, 2023
f25e630
Fix linter false falgs
ChrsMark May 26, 2023
405d58e
Fix typos
ChrsMark May 26, 2023
61f02a1
fix tests
ChrsMark May 26, 2023
bfc694f
make update
ChrsMark May 26, 2023
870aa24
Fix tests
ChrsMark May 29, 2023
d510769
Remove linter skip
ChrsMark May 29, 2023
81408d2
Add changelog
ChrsMark May 29, 2023
22382ca
Merge branch 'main' into rs_watchers
ChrsMark May 29, 2023
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
15 changes: 14 additions & 1 deletion libbeat/autodiscover/providers/kubernetes/pod.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,20 @@ func NewPodEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publish fu
if err != nil {
logger.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
}
metaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, metaConf)
replicaSetWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
logger.Errorf("Error creating watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
}
jobWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_job", client, &kubernetes.Job{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
logger.Errorf("Error creating watcher for %T due to error %+v", &kubernetes.Job{}, err)
}

metaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, replicaSetWatcher, jobWatcher, metaConf)

p := &pod{
config: config,
Expand Down
15 changes: 14 additions & 1 deletion libbeat/processors/add_kubernetes_metadata/kubernetes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,8 +209,21 @@ func (k *kubernetesAnnotator) init(config kubeAnnotatorConfig, cfg *config.C) {
if err != nil {
k.log.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
}
replicaSetWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
k.log.Errorf("Error creating watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
ChrsMark marked this conversation as resolved.
Show resolved Hide resolved
}
jobWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_job", client, &kubernetes.Job{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
k.log.Errorf("Error creating watcher for %T due to error %+v", &kubernetes.Job{}, err)
}

// TODO: refactor the above section to a common function to be used by NeWPodEventer too
metaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, metaConf)
metaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, replicaSetWatcher, jobWatcher, metaConf)

k.indexers = NewIndexers(config.Indexers, metaGen)
k.watcher = watcher
Expand Down
46 changes: 44 additions & 2 deletions metricbeat/module/kubernetes/util/kubernetes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,28 @@ func NewResourceMetadataEnricher(
}
cfg, _ := conf.NewConfigFrom(&commonMetaConfig)

podMetaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, config.AddResourceMetadata)
client, err := kubernetes.GetKubernetesClient(config.KubeConfig, config.KubeClientOptions)
if err != nil {
logp.Err("Error creating Kubernetes client: %s", err)
return &nilEnricher{}
}
replicaSetWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
logp.Err("Error creating watcher for %T due to error %+v", &kubernetes.ReplicaSet{}, err)
return &nilEnricher{}
}

jobWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_job", client, &kubernetes.Job{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
logp.Err("Error creating watcher for %T due to error %+v", &kubernetes.Job{}, err)
return &nilEnricher{}
}

podMetaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, replicaSetWatcher, jobWatcher, config.AddResourceMetadata)

namespaceMeta := metadata.NewNamespaceMetadataGenerator(config.AddResourceMetadata.Namespace, namespaceWatcher.Store(), watcher.Client())
serviceMetaGen := metadata.NewServiceMetadataGenerator(cfg, watcher.Store(), namespaceMeta, watcher.Client())
Expand Down Expand Up @@ -261,14 +282,35 @@ func NewContainerMetadataEnricher(
return &nilEnricher{}
}

client, err := kubernetes.GetKubernetesClient(config.KubeConfig, config.KubeClientOptions)
if err != nil {
logp.Err("Error creating Kubernetes client: %s", err)
return &nilEnricher{}
}
replicaSetWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
logp.Err("Error creating watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
ChrsMark marked this conversation as resolved.
Show resolved Hide resolved
return &nilEnricher{}
}

jobWatcher, err := kubernetes.NewNamedWatcher("resource_metadata_enricher_job", client, &kubernetes.Job{}, kubernetes.WatchOptions{
SyncTimeout: config.SyncPeriod,
}, nil)
if err != nil {
logp.Err("Error creating watcher for %T due to error %+v", &kubernetes.Job{}, err)
return &nilEnricher{}
}

commonMetaConfig := metadata.Config{}
if err := base.Module().UnpackConfig(&commonMetaConfig); err != nil {
logp.Err("Error initializing Kubernetes metadata enricher: %s", err)
return &nilEnricher{}
}
cfg, _ := conf.NewConfigFrom(&commonMetaConfig)

metaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, config.AddResourceMetadata)
metaGen := metadata.GetPodMetaGen(cfg, watcher, nodeWatcher, namespaceWatcher, replicaSetWatcher, jobWatcher, config.AddResourceMetadata)

enricher := buildMetadataEnricher(watcher, nodeWatcher, namespaceWatcher,
// update
Expand Down