admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update dependencies (#6267)

Browse Source 
Signed-off-by: hongming <coder.scala@gmail.com>
(cherry picked from commit cfebd96a1f)
This commit is contained in:
hongming
2025-03-11 14:19:32 +08:00
parent 742c1e52db
commit 39eab5ee5c
4246 changed files with 341171 additions and 131193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
vendor/sigs.k8s.io/controller-runtime/pkg/source/source.go generated vendored
View File

@@ -18,94 +18,174 @@ package source
import (
"context"
"errors"
"fmt"
"sync"
"k8s.io/client-go/util/workqueue"
"k8s.io/utils/ptr"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/handler"
internal "sigs.k8s.io/controller-runtime/pkg/internal/source"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/cache"
"sigs.k8s.io/controller-runtime/pkg/predicate"
)
const (
// defaultBufferSize is the default number of event notifications that can be buffered.
defaultBufferSize = 1024
)
// Source is a source of events (eh.g. Create, Update, Delete operations on Kubernetes Objects, Webhook callbacks, etc)
// Source is a source of events (e.g. Create, Update, Delete operations on Kubernetes Objects, Webhook callbacks, etc)
// which should be processed by event.EventHandlers to enqueue reconcile.Requests.
//
// * Use Kind for events originating in the cluster (e.g. Pod Create, Pod Update, Deployment Update).
//
// * Use Channel for events originating outside the cluster (eh.g. GitHub Webhook callback, Polling external urls).
// * Use Channel for events originating outside the cluster (e.g. GitHub Webhook callback, Polling external urls).
//
// Users may build their own Source implementations.
type Source interface {
// Start is internal and should be called only by the Controller to register an EventHandler with the Informer
// to enqueue reconcile.Requests.
Start(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
type Source = TypedSource[reconcile.Request]
// TypedSource is a generic source of events (e.g. Create, Update, Delete operations on Kubernetes Objects, Webhook callbacks, etc)
// which should be processed by event.EventHandlers to enqueue a request.
//
// * Use Kind for events originating in the cluster (e.g. Pod Create, Pod Update, Deployment Update).
//
// * Use Channel for events originating outside the cluster (e.g. GitHub Webhook callback, Polling external urls).
//
// Users may build their own Source implementations.
type TypedSource[request comparable] interface {
// Start is internal and should be called only by the Controller to start the source.
// Start must be non-blocking.
Start(context.Context, workqueue.TypedRateLimitingInterface[request]) error
}
// SyncingSource is a source that needs syncing prior to being usable. The controller
// will call its WaitForSync prior to starting workers.
type SyncingSource interface {
Source
type SyncingSource = TypedSyncingSource[reconcile.Request]
// TypedSyncingSource is a source that needs syncing prior to being usable. The controller
// will call its WaitForSync prior to starting workers.
type TypedSyncingSource[request comparable] interface {
TypedSource[request]
WaitForSync(ctx context.Context) error
}
// Kind creates a KindSource with the given cache provider.
func Kind(cache cache.Cache, object client.Object) SyncingSource {
return &internal.Kind{Type: object, Cache: cache}
func Kind[object client.Object](
cache cache.Cache,
obj object,
handler handler.TypedEventHandler[object, reconcile.Request],
predicates ...predicate.TypedPredicate[object],
) SyncingSource {
return TypedKind(cache, obj, handler, predicates...)
}
var _ Source = &Channel{}
// TypedKind creates a KindSource with the given cache provider.
func TypedKind[object client.Object, request comparable](
cache cache.Cache,
obj object,
handler handler.TypedEventHandler[object, request],
predicates ...predicate.TypedPredicate[object],
) TypedSyncingSource[request] {
return &internal.Kind[object, request]{
Type: obj,
Cache: cache,
Handler: handler,
Predicates: predicates,
}
}
var _ Source = &channel[string, reconcile.Request]{}
// ChannelOpt allows to configure a source.Channel.
type ChannelOpt[object any, request comparable] func(*channel[object, request])
// WithPredicates adds the configured predicates to a source.Channel.
func WithPredicates[object any, request comparable](p ...predicate.TypedPredicate[object]) ChannelOpt[object, request] {
return func(c *channel[object, request]) {
c.predicates = append(c.predicates, p...)
}
}
// WithBufferSize configures the buffer size for a source.Channel. By
// default, the buffer size is 1024.
func WithBufferSize[object any, request comparable](bufferSize int) ChannelOpt[object, request] {
return func(c *channel[object, request]) {
c.bufferSize = &bufferSize
}
}
// Channel is used to provide a source of events originating outside the cluster
// (e.g. GitHub Webhook callback). Channel requires the user to wire the external
// source (eh.g. http handler) to write GenericEvents to the underlying channel.
type Channel struct {
// source (e.g. http handler) to write GenericEvents to the underlying channel.
func Channel[object any](
source <-chan event.TypedGenericEvent[object],
handler handler.TypedEventHandler[object, reconcile.Request],
opts ...ChannelOpt[object, reconcile.Request],
) Source {
return TypedChannel[object, reconcile.Request](source, handler, opts...)
}
// TypedChannel is used to provide a source of events originating outside the cluster
// (e.g. GitHub Webhook callback). Channel requires the user to wire the external
// source (e.g. http handler) to write GenericEvents to the underlying channel.
func TypedChannel[object any, request comparable](
source <-chan event.TypedGenericEvent[object],
handler handler.TypedEventHandler[object, request],
opts ...ChannelOpt[object, request],
) TypedSource[request] {
c := &channel[object, request]{
source: source,
handler: handler,
}
for _, opt := range opts {
opt(c)
}
return c
}
type channel[object any, request comparable] struct {
// once ensures the event distribution goroutine will be performed only once
once sync.Once
// Source is the source channel to fetch GenericEvents
Source <-chan event.GenericEvent
// source is the source channel to fetch GenericEvents
source <-chan event.TypedGenericEvent[object]
handler handler.TypedEventHandler[object, request]
predicates []predicate.TypedPredicate[object]
bufferSize *int
// dest is the destination channels of the added event handlers
dest []chan event.GenericEvent
// DestBufferSize is the specified buffer size of dest channels.
// Default to 1024 if not specified.
DestBufferSize int
dest []chan event.TypedGenericEvent[object]
// destLock is to ensure the destination channels are safely added/removed
destLock sync.Mutex
}
func (cs *Channel) String() string {
func (cs *channel[object, request]) String() string {
return fmt.Sprintf("channel source: %p", cs)
}
// Start implements Source and should only be called by the Controller.
func (cs *Channel) Start(
func (cs *channel[object, request]) Start(
ctx context.Context,
handler handler.EventHandler,
queue workqueue.RateLimitingInterface,
prct ...predicate.Predicate) error {
queue workqueue.TypedRateLimitingInterface[request],
) error {
// Source should have been specified by the user.
if cs.Source == nil {
if cs.source == nil {
return fmt.Errorf("must specify Channel.Source")
}
// use default value if DestBufferSize not specified
if cs.DestBufferSize == 0 {
cs.DestBufferSize = defaultBufferSize
if cs.handler == nil {
return errors.New("must specify Channel.Handler")
}
dst := make(chan event.GenericEvent, cs.DestBufferSize)
if cs.bufferSize == nil {
cs.bufferSize = ptr.To(1024)
}
dst := make(chan event.TypedGenericEvent[object], *cs.bufferSize)
cs.destLock.Lock()
cs.dest = append(cs.dest, dst)
@@ -119,7 +199,7 @@ func (cs *Channel) Start(
go func() {
for evt := range dst {
shouldHandle := true
for _, p := range prct {
for _, p := range cs.predicates {
if !p.Generic(evt) {
shouldHandle = false
break
@@ -130,7 +210,7 @@ func (cs *Channel) Start(
func() {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
handler.Generic(ctx, evt, queue)
cs.handler.Generic(ctx, evt, queue)
}()
}
}
@@ -139,7 +219,7 @@ func (cs *Channel) Start(
return nil
}
func (cs *Channel) doStop() {
func (cs *channel[object, request]) doStop() {
cs.destLock.Lock()
defer cs.destLock.Unlock()
@@ -148,7 +228,7 @@ func (cs *Channel) doStop() {
}
}
func (cs *Channel) distribute(evt event.GenericEvent) {
func (cs *channel[object, request]) distribute(evt event.TypedGenericEvent[object]) {
cs.destLock.Lock()
defer cs.destLock.Unlock()
@@ -162,14 +242,14 @@ func (cs *Channel) distribute(evt event.GenericEvent) {
}
}
func (cs *Channel) syncLoop(ctx context.Context) {
func (cs *channel[object, request]) syncLoop(ctx context.Context) {
for {
select {
case <-ctx.Done():
// Close destination channels
cs.doStop()
return
case evt, stillOpen := <-cs.Source:
case evt, stillOpen := <-cs.source:
if !stillOpen {
// if the source channel is closed, we're never gonna get
// anything more on it, so stop & bail
@@ -184,21 +264,25 @@ func (cs *Channel) syncLoop(ctx context.Context) {
// Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create).
type Informer struct {
// Informer is the controller-runtime Informer
Informer cache.Informer
Informer cache.Informer
Handler handler.EventHandler
Predicates []predicate.Predicate
}
var _ Source = &Informer{}
// Start is internal and should be called only by the Controller to register an EventHandler with the Informer
// to enqueue reconcile.Requests.
func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface,
prct ...predicate.Predicate) error {
func (is *Informer) Start(ctx context.Context, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
// Informer should have been specified by the user.
if is.Informer == nil {
return fmt.Errorf("must specify Informer.Informer")
}
if is.Handler == nil {
return errors.New("must specify Informer.Handler")
}
_, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, handler, prct).HandlerFuncs())
_, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, is.Handler, is.Predicates).HandlerFuncs())
if err != nil {
return err
}
@@ -212,14 +296,16 @@ func (is *Informer) String() string {
var _ Source = Func(nil)
// Func is a function that implements Source.
type Func func(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
type Func = TypedFunc[reconcile.Request]
// TypedFunc is a function that implements Source.
type TypedFunc[request comparable] func(context.Context, workqueue.TypedRateLimitingInterface[request]) error
// Start implements Source.
func (f Func) Start(ctx context.Context, evt handler.EventHandler, queue workqueue.RateLimitingInterface,
pr ...predicate.Predicate) error {
return f(ctx, evt, queue, pr...)
func (f TypedFunc[request]) Start(ctx context.Context, queue workqueue.TypedRateLimitingInterface[request]) error {
return f(ctx, queue)
}
func (f Func) String() string {
func (f TypedFunc[request]) String() string {
return fmt.Sprintf("func source: %p", f)
}