mirror of
https://github.com/k0sproject/k0s.git
synced 2026-10-28 11:04:40 +08:00
Let the applier manage its own discovery cache (#8305)
This commit is contained in:
10 files changed
+71
-32
No files matched your search
+9
-12
@@ -9,8 +9,6 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/k0sproject/k0s/pkg/kubernetes"
|
||||
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/cli-runtime/pkg/resource"
|
||||
|
||||
@@ -47,12 +45,11 @@ type Applier struct {
|
||||
Name string
|
||||
Dir string
|
||||
|
||||
log *logrus.Entry
|
||||
clientFactory kubernetes.ClientFactoryInterface
|
||||
log *logrus.Entry
|
||||
clients Clients
|
||||
}
|
||||
|
||||
// NewApplier creates new Applier
|
||||
func NewApplier(dir string, kubeClientFactory kubernetes.ClientFactoryInterface) Applier {
|
||||
func NewApplier(dir string, clients Clients) Applier {
|
||||
name := filepath.Base(dir)
|
||||
log := logrus.WithFields(logrus.Fields{
|
||||
"component": "applier",
|
||||
@@ -60,10 +57,10 @@ func NewApplier(dir string, kubeClientFactory kubernetes.ClientFactoryInterface)
|
||||
})
|
||||
|
||||
return Applier{
|
||||
log: log,
|
||||
Dir: dir,
|
||||
Name: name,
|
||||
clientFactory: kubeClientFactory,
|
||||
log: log,
|
||||
Dir: dir,
|
||||
Name: name,
|
||||
clients: clients,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -81,7 +78,7 @@ func (a *Applier) Apply(ctx context.Context) error {
|
||||
stack := Stack{
|
||||
Name: a.Name,
|
||||
Resources: resources,
|
||||
Clients: a.clientFactory,
|
||||
Clients: a.clients,
|
||||
}
|
||||
a.log.Debug("applying stack")
|
||||
err = stack.Apply(ctx, true)
|
||||
@@ -96,7 +93,7 @@ func (a *Applier) Apply(ctx context.Context) error {
|
||||
|
||||
// Delete deletes the entire stack by applying it with empty set of resources
|
||||
func (a *Applier) Delete(ctx context.Context) error {
|
||||
stack := Stack{Name: a.Name, Clients: a.clientFactory}
|
||||
stack := Stack{Name: a.Name, Clients: a.clients}
|
||||
logrus.Debugf("about to delete a stack %s with empty apply", a.Name)
|
||||
return stack.Apply(ctx, true)
|
||||
}
|
||||
|
||||
@@ -76,7 +76,8 @@ spec:
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "test-deploy.yaml"), []byte(templateDeployment), 0400))
|
||||
|
||||
fakes := kubeutil.NewFakeClientFactory()
|
||||
a := applier.NewApplier(dir, fakes)
|
||||
clients := applier.NewClients(fakes)
|
||||
a := applier.NewApplier(dir, clients)
|
||||
|
||||
ctx := t.Context()
|
||||
err := a.Apply(ctx)
|
||||
@@ -97,7 +98,7 @@ spec:
|
||||
assert.NoError(t, err)
|
||||
|
||||
// Attempt to delete the stack with a different applier
|
||||
a2 := applier.NewApplier(dir, fakes)
|
||||
a2 := applier.NewApplier(dir, clients)
|
||||
assert.NoError(t, a2.Delete(ctx))
|
||||
// Check that the resources are deleted
|
||||
_, err = fakes.DynamicClient.Resource(*gv).Namespace(metav1.NamespaceSystem).Get(ctx, "applier-test", metav1.GetOptions{})
|
||||
|
||||
@@ -120,9 +120,10 @@ func (m *Manager) runWatchers(ctx context.Context) {
|
||||
ctx, cancel := context.WithCancelCause(ctx)
|
||||
defer cancel(nil) // satisfy linter, not required for correctness
|
||||
stacks := make(map[string]stack, len(dirs))
|
||||
clients := NewClients(m.KubeClientFactory)
|
||||
|
||||
for _, dir := range dirs {
|
||||
m.createStack(ctx, stacks, filepath.Join(m.bundleDir, dir))
|
||||
m.createStack(ctx, stacks, clients, filepath.Join(m.bundleDir, dir))
|
||||
}
|
||||
|
||||
for {
|
||||
@@ -135,7 +136,7 @@ func (m *Manager) runWatchers(ctx context.Context) {
|
||||
switch event.Op {
|
||||
case fsnotify.Create:
|
||||
if dir.IsDirectory(event.Name) {
|
||||
m.createStack(ctx, stacks, event.Name)
|
||||
m.createStack(ctx, stacks, clients, event.Name)
|
||||
}
|
||||
case fsnotify.Remove:
|
||||
m.removeStack(ctx, stacks, event.Name)
|
||||
@@ -152,7 +153,7 @@ func (m *Manager) runWatchers(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) createStack(ctx context.Context, stacks map[string]stack, name string) {
|
||||
func (m *Manager) createStack(ctx context.Context, stacks map[string]stack, clients Clients, name string) {
|
||||
// safeguard in case the fswatcher would trigger an event for an already existing stack
|
||||
if _, ok := stacks[name]; ok {
|
||||
return
|
||||
@@ -174,7 +175,7 @@ func (m *Manager) createStack(ctx context.Context, stacks map[string]stack, name
|
||||
ctx, cancel := context.WithCancelCause(ctx)
|
||||
stopped := make(chan struct{})
|
||||
|
||||
stack := stack{cancel, stopped, NewStackApplier(name, m.KubeClientFactory)}
|
||||
stack := stack{cancel, stopped, NewStackApplier(name, clients)}
|
||||
stacks[name] = stack
|
||||
|
||||
go func() {
|
||||
|
||||
+44
-4
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/k0sproject/k0s/pkg/kubernetes/watch"
|
||||
|
||||
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
|
||||
apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
|
||||
apiErrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
@@ -25,6 +26,7 @@ import (
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/client-go/discovery"
|
||||
"k8s.io/client-go/discovery/cached/memory"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/restmapper"
|
||||
|
||||
@@ -33,14 +35,52 @@ import (
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// Provides the Kubernetes clients required to apply a [Stack].
|
||||
type Clients interface {
|
||||
GetDiscoveryClient() (discovery.CachedDiscoveryInterface, error)
|
||||
GetDynamicClient() (dynamic.Interface, error)
|
||||
GetAPIExtensionsClient() (apiextensionsclientset.Interface, error)
|
||||
}
|
||||
|
||||
// Returns a [Clients] implementation backed by clientFactory. It lazily creates
|
||||
// a memory-cached discovery client that lives as long as the returned instance.
|
||||
// Hence the lifetime of the returned instance determines the lifetime of the
|
||||
// discovery cache.
|
||||
func NewClients(clientFactory kubernetes.ClientFactoryInterface) Clients {
|
||||
return &clients{ClientFactoryInterface: clientFactory}
|
||||
}
|
||||
|
||||
type clients struct {
|
||||
kubernetes.ClientFactoryInterface
|
||||
mu sync.Mutex
|
||||
discovery discovery.CachedDiscoveryInterface
|
||||
}
|
||||
|
||||
// GetDiscoveryClient implements [Clients].
|
||||
func (c *clients) GetDiscoveryClient() (discovery.CachedDiscoveryInterface, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if c.discovery == nil {
|
||||
client, err := c.GetClient()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.discovery = memory.NewMemCacheClient(client.Discovery())
|
||||
}
|
||||
|
||||
return c.discovery, nil
|
||||
}
|
||||
|
||||
func ApplyStack(ctx context.Context, clients kubernetes.ClientFactoryInterface, resources []*unstructured.Unstructured, stackName string) error {
|
||||
applierClients := NewClients(clients)
|
||||
var lastErr error
|
||||
if err := retry.Do(
|
||||
func() error {
|
||||
stack := Stack{
|
||||
Name: stackName,
|
||||
Resources: resources,
|
||||
Clients: clients,
|
||||
Clients: applierClients,
|
||||
}
|
||||
lastErr = stack.Apply(ctx, true)
|
||||
return lastErr
|
||||
@@ -66,7 +106,7 @@ type Stack struct {
|
||||
Name string
|
||||
Resources []*unstructured.Unstructured
|
||||
keepResources []string
|
||||
Clients kubernetes.ClientFactoryInterface
|
||||
Clients Clients
|
||||
|
||||
log *logrus.Entry
|
||||
}
|
||||
@@ -76,11 +116,11 @@ type Stack struct {
|
||||
func (s *Stack) Apply(ctx context.Context, prune bool) error {
|
||||
s.log = logrus.WithField("stack", s.Name)
|
||||
|
||||
discoveryClient, err := s.Clients.GetDiscoveryClient()
|
||||
dynamicClient, err := s.Clients.GetDynamicClient()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dynamicClient, err := s.Clients.GetDynamicClient()
|
||||
discoveryClient, err := s.Clients.GetDiscoveryClient()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -34,7 +34,7 @@ func TestStack_StripsNamespaceFromClusterScopedResource(t *testing.T) {
|
||||
s := applier.Stack{
|
||||
Name: "strip-ns",
|
||||
Resources: []*unstructured.Unstructured{resources},
|
||||
Clients: fakes,
|
||||
Clients: applier.NewClients(fakes),
|
||||
}
|
||||
|
||||
err = s.Apply(t.Context(), true)
|
||||
@@ -61,11 +61,12 @@ func TestStack_StackNameChangeTriggersUpdateWithoutChecksumChange(t *testing.T)
|
||||
require.NoError(t, err)
|
||||
|
||||
fakes := testutil.NewFakeClientFactory()
|
||||
clients := applier.NewClients(fakes)
|
||||
|
||||
oldStack := applier.Stack{
|
||||
Name: "old",
|
||||
Resources: []*unstructured.Unstructured{resources},
|
||||
Clients: fakes,
|
||||
Clients: clients,
|
||||
}
|
||||
require.NoError(t, oldStack.Apply(t.Context(), true))
|
||||
|
||||
@@ -80,7 +81,7 @@ func TestStack_StackNameChangeTriggersUpdateWithoutChecksumChange(t *testing.T)
|
||||
newStack := applier.Stack{
|
||||
Name: "new",
|
||||
Resources: []*unstructured.Unstructured{resources},
|
||||
Clients: fakes,
|
||||
Clients: clients,
|
||||
}
|
||||
require.NoError(t, newStack.Apply(t.Context(), false))
|
||||
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/avast/retry-go"
|
||||
"github.com/k0sproject/k0s/pkg/kubernetes"
|
||||
|
||||
"github.com/fsnotify/fsnotify"
|
||||
"github.com/sirupsen/logrus"
|
||||
@@ -27,9 +26,9 @@ type StackApplier struct {
|
||||
}
|
||||
|
||||
// NewStackApplier crates new stack applier to manage a stack
|
||||
func NewStackApplier(path string, kubeClientFactory kubernetes.ClientFactoryInterface) *StackApplier {
|
||||
func NewStackApplier(path string, clients Clients) *StackApplier {
|
||||
var mu sync.Mutex
|
||||
applier := NewApplier(path, kubeClientFactory)
|
||||
applier := NewApplier(path, clients)
|
||||
|
||||
return &StackApplier{
|
||||
log: logrus.WithField("component", "applier-"+applier.Name),
|
||||
|
||||
@@ -120,7 +120,7 @@ func (a *Autopilot) applyManifests(ctx context.Context) {
|
||||
stack := applier.Stack{
|
||||
Name: AutopilotStackName,
|
||||
Resources: slices.Concat(crdResources, stackResources),
|
||||
Clients: a.AdminClientFactory,
|
||||
Clients: applier.NewClients(a.AdminClientFactory),
|
||||
}
|
||||
|
||||
for {
|
||||
|
||||
@@ -240,7 +240,7 @@ func (i *ClusterConfigInitializer) applyAPIConfigStack(ctx context.Context) erro
|
||||
return (&applier.Stack{
|
||||
Name: ClusterConfigStackName,
|
||||
Resources: []*unstructured.Unstructured{&crd},
|
||||
Clients: i.clients,
|
||||
Clients: applier.NewClients(i.clients),
|
||||
}).Apply(ctx, true)
|
||||
}
|
||||
|
||||
|
||||
@@ -80,7 +80,7 @@ func (c *CRDStack) Start(context.Context) error {
|
||||
stack := applier.Stack{
|
||||
Name: c.stackName,
|
||||
Resources: resources,
|
||||
Clients: c.clients,
|
||||
Clients: applier.NewClients(c.clients),
|
||||
}
|
||||
|
||||
for {
|
||||
|
||||
@@ -128,7 +128,7 @@ func (r *Reconciler) Init(context.Context) error {
|
||||
apply := func(ctx context.Context, resources resources) error {
|
||||
return (&applier.Stack{
|
||||
Name: fmt.Sprintf("k0s-%s-%s", constant.WorkerConfigComponentName, constant.KubernetesMajorMinorVersion),
|
||||
Clients: clientFactory,
|
||||
Clients: applier.NewClients(clientFactory),
|
||||
Resources: resources,
|
||||
}).Apply(ctx, true)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user