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

feat: setup a resync period for shared informers #1090

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: setup a resync period for shared informers #1090

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
29 changes: 25 additions & 4 deletions scrapers/kubernetes/informers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"fmt"
"strings"
"sync"
"time"

"github.com/flanksource/commons/logger"
"github.com/flanksource/config-db/api"
Expand Down Expand Up @@ -37,7 +38,12 @@ type SharedInformerManager struct {

type DeleteObjHandler func(ctx context.Context, id string) error

func (t *SharedInformerManager) Register(ctx api.ScrapeContext, watchResource v1.KubernetesResourceToWatch, buffer chan<- *unstructured.Unstructured, deleteBuffer chan<- string) error {
func (t *SharedInformerManager) Register(
ctx api.ScrapeContext,
watchResource v1.KubernetesResourceToWatch,
buffer chan<- *unstructured.Unstructured,
deleteBuffer chan<- string,
) error {
apiVersion, kind := watchResource.ApiVersion, watchResource.Kind

informer, stopper, isNew := t.getOrCreate(ctx, apiVersion, kind)
Expand Down Expand Up @@ -82,7 +88,13 @@ func (t *SharedInformerManager) Register(ctx api.ScrapeContext, watchResource v1
DeleteFunc: func(obj any) {
u, err := getUnstructuredFromInformedObj(watchResource, obj)
if err != nil {
logToJobHistory(ctx.DutyContext(), "DeleteK8sWatchResource", ctx.ScrapeConfig().GetPersistedID(), "failed to get unstructured %v", err)
logToJobHistory(
ctx.DutyContext(),
"DeleteK8sWatchResource",
ctx.ScrapeConfig().GetPersistedID(),
"failed to get unstructured %v",
err,
)
return
}

Expand All @@ -104,7 +116,10 @@ func (t *SharedInformerManager) Register(ctx api.ScrapeContext, watchResource v1
}

// getOrCreate returns an existing shared informer instance or creates & returns a new one.
func (t *SharedInformerManager) getOrCreate(ctx api.ScrapeContext, apiVersion, kind string) (informers.GenericInformer, chan struct{}, bool) {
func (t *SharedInformerManager) getOrCreate(
ctx api.ScrapeContext,
apiVersion, kind string,
) (informers.GenericInformer, chan struct{}, bool) {
t.mu.Lock()
defer t.mu.Unlock()

Expand All @@ -117,7 +132,13 @@ func (t *SharedInformerManager) getOrCreate(ctx api.ScrapeContext, apiVersion, k
}
}

factory := informers.NewSharedInformerFactory(ctx.Kubernetes(), 0)
// The resync period is chosen based off of the sample controller in here
// https://github.com/kubernetes/sample-controller/blob/d90ced32cf661b9821ecb20b08502ad62377a035/main.go#L63
//
// The sample uses 30s as the resync period, however we're using 5m in here to reduce the load on the API server.
resyncPeriod := time.Minute * 5

factory := informers.NewSharedInformerFactory(ctx.Kubernetes(), resyncPeriod)
stopper := make(chan struct{})

informer, err := getInformer(factory, apiVersion, kind)
Expand Down
Loading