github/client-go
1
0
Fork 0
mirror of https://github.com/kubernetes/client-go.git synced 2026-08-09 16:01:07 +00:00
Code Issues Projects Releases Wiki Activity

Revert "Merge pull request #140448 from xigang/event_broadcaster_goroutine_leak"

Browse Source 
This reverts commit 112ff970abd6f79f23486550cf15deedab5816ef, reversing
changes made to c1c118b86f31187bd918cfe699c139fa4b5050b2.

Kubernetes-commit: eb00f13d4abd2f446b9bdb4e7aec0c4c3e1b786d
This commit is contained in:
xigang
2026-07-28 22:49:22 +08:00
committed by Kubernetes Publisher
parent df92378354
commit 7b892feea6
2 changed files with 53 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
tools/events/event_broadcaster.go
View File

@@ -25,7 +25,6 @@ import (
corev1 "k8s.io/api/core/v1"
eventsv1 "k8s.io/api/events/v1"
apiequality "k8s.io/apimachinery/pkg/api/equality"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
@@ -51,7 +50,6 @@ const (
finishTime = 6 * time.Minute
refreshTime = 30 * time.Minute
maxQueuedEvents = 1000
recordWorkers = 8
)
var defaultSleepDuration = 10 * time.Second
@@ -73,8 +71,6 @@ type eventBroadcasterImpl struct {
eventCache map[eventKey]*eventsv1.Event
sleepDuration time.Duration
sink EventSink
eventQueue chan *eventsv1.Event
cancel func()
}
// EventSinkImpl wraps EventsV1Interface to implement EventSink.
@@ -120,94 +116,48 @@ func newBroadcaster(sink EventSink, sleepDuration time.Duration, eventCache map[
eventCache: eventCache,
sleepDuration: sleepDuration,
sink: sink,
eventQueue: make(chan *eventsv1.Event, maxQueuedEvents),
}
}
func (e *eventBroadcasterImpl) Shutdown() {
e.mu.Lock()
if e.cancel != nil {
e.cancel()
e.cancel = nil
}
e.mu.Unlock()
e.Broadcaster.Shutdown()
}
// refreshExistingEventSeries refresh events TTL
func (e *eventBroadcasterImpl) refreshExistingEventSeries(ctx context.Context) {
type eventSnapshot struct {
key eventKey
event *eventsv1.Event
}
var snapshots []eventSnapshot
// TODO: Investigate whether lock contention won't be a problem
e.mu.Lock()
defer e.mu.Unlock()
for isomorphicKey, event := range e.eventCache {
if event.Series != nil {
snapshots = append(snapshots, eventSnapshot{key: isomorphicKey, event: event.DeepCopy()})
if recordedEvent, retry := recordEvent(ctx, e.sink, event); !retry {
if recordedEvent != nil {
e.eventCache[isomorphicKey] = recordedEvent
}
}
}
}
e.mu.Unlock()
for _, snapshot := range snapshots {
recordedEvent, retry := recordEvent(ctx, e.sink, snapshot.event.DeepCopy())
if retry || recordedEvent == nil {
continue
}
e.mu.Lock()
cachedEvent, exists := e.eventCache[snapshot.key]
// The sink call ran without the lock. Do not overwrite counts or
// timestamps which were aggregated while it was in flight.
if exists && apiequality.Semantic.DeepEqual(cachedEvent, snapshot.event) {
e.eventCache[snapshot.key] = recordedEvent
}
e.mu.Unlock()
}
}
// finishSeries checks if a series has ended and either:
// - write final count to the apiserver
// - delete a singleton event (i.e. series field is nil) from the cache
func (e *eventBroadcasterImpl) finishSeries(ctx context.Context) {
type eventSnapshot struct {
key eventKey
event *eventsv1.Event
shouldRecord bool
}
now := time.Now()
var snapshots []eventSnapshot
// TODO: Investigate whether lock contention won't be a problem
e.mu.Lock()
defer e.mu.Unlock()
for isomorphicKey, event := range e.eventCache {
eventSerie := event.Series
if eventSerie != nil {
if eventSerie.LastObservedTime.Time.Before(now.Add(-finishTime)) {
snapshots = append(snapshots, eventSnapshot{key: isomorphicKey, event: event.DeepCopy(), shouldRecord: true})
if eventSerie.LastObservedTime.Time.Before(time.Now().Add(-finishTime)) {
if _, retry := recordEvent(ctx, e.sink, event); !retry {
delete(e.eventCache, isomorphicKey)
}
}
} else if event.EventTime.Time.Before(now.Add(-finishTime)) {
snapshots = append(snapshots, eventSnapshot{key: isomorphicKey, event: event.DeepCopy()})
} else if event.EventTime.Time.Before(time.Now().Add(-finishTime)) {
delete(e.eventCache, isomorphicKey)
}
}
e.mu.Unlock()
for _, snapshot := range snapshots {
if snapshot.shouldRecord {
if _, retry := recordEvent(ctx, e.sink, snapshot.event.DeepCopy()); retry {
continue
}
}
e.mu.Lock()
cachedEvent, exists := e.eventCache[snapshot.key]
// An event observed while the sink call was in flight means the series
// is active again and must remain in the cache.
if exists && apiequality.Semantic.DeepEqual(cachedEvent, snapshot.event) {
delete(e.eventCache, snapshot.key)
}
e.mu.Unlock()
}
}
// NewRecorder returns an EventRecorder that records events with the given event source.
@@ -220,52 +170,39 @@ func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, reportingCont
func (e *eventBroadcasterImpl) recordToSink(ctx context.Context, event *eventsv1.Event, clock clock.Clock) {
// Make a copy before modification, because there could be multiple listeners.
eventCopy := event.DeepCopy()
record := func() *eventsv1.Event {
e.mu.Lock()
defer e.mu.Unlock()
eventKey := getKey(eventCopy)
isomorphicEvent, isIsomorphic := e.eventCache[eventKey]
if isIsomorphic {
if isomorphicEvent.Series != nil {
isomorphicEvent.Series.Count++
isomorphicEvent.Series.LastObservedTime = metav1.MicroTime{Time: clock.Now()}
return nil
go func() {
evToRecord := func() *eventsv1.Event {
e.mu.Lock()
defer e.mu.Unlock()
eventKey := getKey(eventCopy)
isomorphicEvent, isIsomorphic := e.eventCache[eventKey]
if isIsomorphic {
if isomorphicEvent.Series != nil {
isomorphicEvent.Series.Count++
isomorphicEvent.Series.LastObservedTime = metav1.MicroTime{Time: clock.Now()}
return nil
}
isomorphicEvent.Series = &eventsv1.EventSeries{
Count: 2,
LastObservedTime: metav1.MicroTime{Time: clock.Now()},
}
// Make a copy of the Event to make sure that recording it
// doesn't mess with the object stored in cache.
return isomorphicEvent.DeepCopy()
}
isomorphicEvent.Series = &eventsv1.EventSeries{
Count: 2,
LastObservedTime: metav1.MicroTime{Time: clock.Now()},
}
// Make a copy of the Event to make sure that recording it
// doesn't mess with the object stored in cache.
return isomorphicEvent.DeepCopy()
}
e.eventCache[eventKey] = eventCopy
// Make a copy of the Event to make sure that recording it doesn't
// mess with the object stored in cache.
return eventCopy.DeepCopy()
}()
if record != nil {
select {
case e.eventQueue <- record:
default:
klog.FromContext(ctx).Error(nil, "Unable to record event: too many queued events, dropped event", "event", record)
}
}
}
// recordingWorker delivers queued events to the sink until the context is
// canceled.
func (e *eventBroadcasterImpl) recordingWorker(ctx context.Context) {
defer utilruntime.HandleCrash()
for {
select {
case <-ctx.Done():
return
case record := <-e.eventQueue:
e.eventCache[eventKey] = eventCopy
// Make a copy of the Event to make sure that recording it doesn't
// mess with the object stored in cache.
return eventCopy.DeepCopy()
}()
if evToRecord != nil {
// TODO: Add a metric counting the number of recording attempts
e.attemptRecording(ctx, record)
e.attemptRecording(ctx, evToRecord)
// We don't want the new recorded Event to be reflected in the
// client's cache because server-side mutations could mess with the
// aggregation mechanism used by the client.
}
}
}()
}
func (e *eventBroadcasterImpl) attemptRecording(ctx context.Context, event *eventsv1.Event) {
@@ -439,9 +376,6 @@ func (e *eventBroadcasterImpl) startRecordingEvents(ctx context.Context) error {
if err != nil {
return err
}
for range recordWorkers {
go e.recordingWorker(ctx)
}
go func() {
<-ctx.Done()
stopWatcher()
@@ -460,20 +394,9 @@ func (e *eventBroadcasterImpl) StartRecordingToSink(stopCh <-chan struct{}) {
// StartRecordingToSinkWithContext starts sending events received from the specified eventBroadcaster to the given sink.
func (e *eventBroadcasterImpl) StartRecordingToSinkWithContext(ctx context.Context) error {
e.mu.Lock()
defer e.mu.Unlock()
if e.cancel != nil {
return fmt.Errorf("broadcaster is already recording to a sink")
}
cctx, cancel := context.WithCancel(ctx)
if err := e.startRecordingEvents(cctx); err != nil {
cancel()
return err
}
go wait.UntilWithContext(cctx, e.refreshExistingEventSeries, refreshTime)
go wait.UntilWithContext(cctx, e.finishSeries, finishTime)
e.cancel = cancel
return nil
go wait.UntilWithContext(ctx, e.refreshExistingEventSeries, refreshTime)
go wait.UntilWithContext(ctx, e.finishSeries, finishTime)
return e.startRecordingEvents(ctx)
}
type eventBroadcasterAdapterImpl struct {

13
tools/events/event_recorder.go
View File

@@ -24,6 +24,7 @@ import (
eventsv1 "k8s.io/api/events/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/record/util"
"k8s.io/client-go/tools/reference"
@@ -90,14 +91,10 @@ func (recorder *recorderImpl) eventf(logger klog.Logger, regarding runtime.Objec
return
}
event := recorder.makeEvent(refRegarding, refRelated, timestamp, annotations, eventtype, reason, message, recorder.reportingController, recorder.reportingInstance, action)
sent, err := recorder.ActionOrDrop(watch.Added, event)
if err != nil {
logger.Error(err, "Unable to record event (will not retry!)")
return
}
if !sent {
logger.Error(nil, "Unable to record event: too many queued events, dropped event", "event", event)
}
go func() {
defer utilruntime.HandleCrash()
recorder.Action(watch.Added, event)
}()
}
func (recorder *recorderImpl) makeEvent(refRegarding *v1.ObjectReference, refRelated *v1.ObjectReference, timestamp metav1.MicroTime, annotations map[string]string, eventtype, reason, message string, reportingController string, reportingInstance string, action string) *eventsv1.Event {