func (c *EventCorrelator) EventCorrelate(newEvent *api.Event) (*EventCorrelateResult, error) {
	if c.filterFunc(newEvent) {
		return &EventCorrelateResult{Skip: true}, nil
	}
	aggregateEvent, err := c.aggregator.EventAggregate(newEvent)
	if err != nil {
		return &EventCorrelateResult{}, err
	}
	observedEvent, patch, err := c.logger.eventObserve(aggregateEvent)
	return &EventCorrelateResult{Event: observedEvent, Patch: patch}, err
}
EventCorrelate filters, aggregates, counts, and de-duplicates all incoming events
package record
import (
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"
	"github.com/golang/groupcache/lru"
	"k8s.io/kubernetes/pkg/api"
	"k8s.io/kubernetes/pkg/api/unversioned"
	"k8s.io/kubernetes/pkg/util"
	"k8s.io/kubernetes/pkg/util/sets"
	"k8s.io/kubernetes/pkg/util/strategicpatch"
)
const (
	maxLruCacheEntries = 4096
	defaultAggregateMaxEvents         = 10
	defaultAggregateIntervalInSeconds = 600
)
func getEventKey(event *api.Event) string {
	return strings.Join([]string{
		event.Source.Component,
		event.Source.Host,
		event.InvolvedObject.Kind,
		event.InvolvedObject.Namespace,
		event.InvolvedObject.Name,
		string(event.InvolvedObject.UID),
		event.InvolvedObject.APIVersion,
		event.Type,
		event.Reason,
		event.Message,
	},
		"")
}
type EventFilterFunc func(event *api.Event) bool
func DefaultEventFilterFunc(event *api.Event) bool {
	return false
}
type EventAggregatorKeyFunc func(event *api.Event) (aggregateKey string, localKey string)
func EventAggregatorByReasonFunc(event *api.Event) (string, string) {
	return strings.Join([]string{
		event.Source.Component,
		event.Source.Host,
		event.InvolvedObject.Kind,
		event.InvolvedObject.Namespace,
		event.InvolvedObject.Name,
		string(event.InvolvedObject.UID),
		event.InvolvedObject.APIVersion,
		event.Type,
		event.Reason,
	},
		""), event.Message
}
type EventAggregatorMessageFunc func(event *api.Event) string
func EventAggregatorByReasonMessageFunc(event *api.Event) string {
	return "(events with common reason combined)"
}
type EventAggregator struct {
	sync.RWMutex
	cache *lru.Cache
	keyFunc EventAggregatorKeyFunc
	messageFunc EventAggregatorMessageFunc
	maxEvents int
	maxIntervalInSeconds int
	clock util.Clock
}
func NewEventAggregator(lruCacheSize int, keyFunc EventAggregatorKeyFunc, messageFunc EventAggregatorMessageFunc,
	maxEvents int, maxIntervalInSeconds int, clock util.Clock) *EventAggregator {
	return &EventAggregator{
		cache:                lru.New(lruCacheSize),
		keyFunc:              keyFunc,
		messageFunc:          messageFunc,
		maxEvents:            maxEvents,
		maxIntervalInSeconds: maxIntervalInSeconds,
		clock:                clock,
	}
}
type aggregateRecord struct {
	localKeys sets.String
	lastTimestamp unversioned.Time
}
func (e *EventAggregator) EventAggregate(newEvent *api.Event) (*api.Event, error) {
	aggregateKey, localKey := e.keyFunc(newEvent)
	now := unversioned.NewTime(e.clock.Now())
	record := aggregateRecord{localKeys: sets.NewString(), lastTimestamp: now}
	e.Lock()
	defer e.Unlock()
	value, found := e.cache.Get(aggregateKey)
	if found {
		record = value.(aggregateRecord)
	}
	maxInterval := time.Duration(e.maxIntervalInSeconds) * time.Second
	interval := now.Time.Sub(record.lastTimestamp.Time)
	if interval > maxInterval {
		record = aggregateRecord{localKeys: sets.NewString()}
	}
	record.localKeys.Insert(localKey)
	record.lastTimestamp = now
	e.cache.Add(aggregateKey, record)
	if record.localKeys.Len() < e.maxEvents {
		return newEvent, nil
	}
	record.localKeys.PopAny()
	eventCopy := &api.Event{
		ObjectMeta: api.ObjectMeta{
			Name:      fmt.Sprintf("%v.%x", newEvent.InvolvedObject.Name, now.UnixNano()),
			Namespace: newEvent.Namespace,
		},
		Count:          1,
		FirstTimestamp: now,
		InvolvedObject: newEvent.InvolvedObject,
		LastTimestamp:  now,
		Message:        e.messageFunc(newEvent),
		Type:           newEvent.Type,
		Reason:         newEvent.Reason,
		Source:         newEvent.Source,
	}
	return eventCopy, nil
}
type eventLog struct {
	count int
	firstTimestamp unversioned.Time
	name string
	resourceVersion string
}
type eventLogger struct {
	sync.RWMutex
	cache *lru.Cache
	clock util.Clock
}
func newEventLogger(lruCacheEntries int, clock util.Clock) *eventLogger {
	return &eventLogger{cache: lru.New(lruCacheEntries), clock: clock}
}
func (e *eventLogger) eventObserve(newEvent *api.Event) (*api.Event, []byte, error) {
	var (
		patch []byte
		err   error
	)
	key := getEventKey(newEvent)
	eventCopy := *newEvent
	event := &eventCopy
	e.Lock()
	defer e.Unlock()
	lastObservation := e.lastEventObservationFromCache(key)
	if lastObservation.count > 0 {
		event.Name = lastObservation.name
		event.ResourceVersion = lastObservation.resourceVersion
		event.FirstTimestamp = lastObservation.firstTimestamp
		event.Count = lastObservation.count + 1
		eventCopy2 := *event
		eventCopy2.Count = 0
		eventCopy2.LastTimestamp = unversioned.NewTime(time.Unix(0, 0))
		newData, _ := json.Marshal(event)
		oldData, _ := json.Marshal(eventCopy2)
		patch, err = strategicpatch.CreateStrategicMergePatch(oldData, newData, event)
	}
	e.cache.Add(
		key,
		eventLog{
			count:           event.Count,
			firstTimestamp:  event.FirstTimestamp,
			name:            event.Name,
			resourceVersion: event.ResourceVersion,
		},
	)
	return event, patch, err
}
func (e *eventLogger) updateState(event *api.Event) {
	key := getEventKey(event)
	e.Lock()
	defer e.Unlock()
	e.cache.Add(
		key,
		eventLog{
			count:           event.Count,
			firstTimestamp:  event.FirstTimestamp,
			name:            event.Name,
			resourceVersion: event.ResourceVersion,
		},
	)
}
func (e *eventLogger) lastEventObservationFromCache(key string) eventLog {
	value, ok := e.cache.Get(key)
	if ok {
		observationValue, ok := value.(eventLog)
		if ok {
			return observationValue
		}
	}
	return eventLog{}
}
type EventCorrelator struct {
	filterFunc EventFilterFunc
	aggregator *EventAggregator
	logger *eventLogger
}
type EventCorrelateResult struct {
	Event *api.Event
	Patch []byte
	Skip bool
}
func NewEventCorrelator(clock util.Clock) *EventCorrelator {
	cacheSize := maxLruCacheEntries
	return &EventCorrelator{
		filterFunc: DefaultEventFilterFunc,
		aggregator: NewEventAggregator(
			cacheSize,
			EventAggregatorByReasonFunc,
			EventAggregatorByReasonMessageFunc,
			defaultAggregateMaxEvents,
			defaultAggregateIntervalInSeconds,
			clock),
		logger: newEventLogger(cacheSize, clock),
	}
}