events

Documentation

Index

• type Config
• type Event
• type EventAccumulator
  • func (e *EventAccumulator) Add(event Event) int
  • func (e *EventAccumulator) Flush(userManifestType UserManifestType) EventJsons
  • func (e *EventAccumulator) FlushAll() map[UserManifestType]EventJsons
  • func (e *EventAccumulator) GetKeys() []UserManifestType
  • func (e *EventAccumulator) IsEmpty(userManifestType UserManifestType) bool
• type EventEngine
  • func NewEventEngine(ctx context.Context, log logr.Logger, config *Config, client client.Client) *EventEngine
  • func (e *EventEngine) IsReady() bool
  • func (e *EventEngine) SetKubeClient(client client.Client)
  • func (e *EventEngine) Start(ctx context.Context) error
• type EventJsons
• type EventProcessorSender
• type EventReporter
  • func NewEventReporter(log logr.Logger, config *Config, dataService *dataservice.DataService) (*EventReporter, error)
  • func (r *EventReporter) Report(metadata Metadata, manifestType string, eventJsons EventJsons) error
• type Events
• type Manifest
• type Metadata
• type ProcessorSender
  • func (p *ProcessorSender) Process(ctx context.Context, event Event) error
  • func (p *ProcessorSender) ReportOne(event Event) error
  • func (p *ProcessorSender) Send(ctx context.Context, userManifestType UserManifestType) error
  • func (p *ProcessorSender) SendAll(ctx context.Context) error
  • func (p *ProcessorSender) Start(ctx context.Context, ready chan bool) error
  • func (p *ProcessorSender) UpdateErrorStatus(ctx context.Context) error
• type ReportData
• type TarGzipPool
  • func (pool *TarGzipPool) TarGzip(src string, dest string) error
• type UserConfigs
  • func (a *UserConfigs) GetMetadata(userName string) Metadata
  • func (a *UserConfigs) HasUser(userName string) bool
  • func (a *UserConfigs) SetUserConfigs(userConfigs []v1alpha1.UserConfig)
• type UserManifestType

Types

type Config

type Config struct {
	OutputDirectory      string
	DataServiceTokenFile string
	DataServiceCertFile  string
	DataServiceURL       *url.URL
	Namespace            string
	UserConfigs          UserConfigs
	AccMemoryLimit       resource.Quantity
	MaxFlushTimeout      metav1.Duration
	MaxEventEntries      int
	LicenseAccept        bool
	CipherSuites         []uint16
	MinVersion           uint16
}

config needed to provideDataServiceConfig

type Event

type Event struct {
	User string // x-remote-user the Event was posted by
	json.RawMessage
	ManifestType string // dataReporter or override
}

type EventAccumulator

type EventAccumulator struct {
	// contains filtered or unexported fields
}

EventAccumulator collects events associated with a User and ManifestType [User][ManifestType]EventJsons

func (*EventAccumulator) Add

func (e *EventAccumulator) Add(event Event) int

Add an Event for a [User][ManifestType]

func (*EventAccumulator) Flush

func (e *EventAccumulator) Flush(userManifestType UserManifestType) EventJsons

Flush all Events for a [User][ManifestType]

func (*EventAccumulator) FlushAll

func (e *EventAccumulator) FlushAll() map[UserManifestType]EventJsons

Flush EventMap and reset to clear memory accumulation

func (*EventAccumulator) GetKeys

func (e *EventAccumulator) GetKeys() []UserManifestType

Get the userManifestType keys

func (*EventAccumulator) IsEmpty

func (e *EventAccumulator) IsEmpty(userManifestType UserManifestType) bool

Are the events empty for a [User][ManifestType]

type EventEngine

type EventEngine struct {
	*ProcessorSender
	// contains filtered or unexported fields
}

func NewEventEngine

func NewEventEngine(
	ctx context.Context,
	log logr.Logger,
	config *Config,
	client client.Client,
) *EventEngine

func (*EventEngine) IsReady

func (e *EventEngine) IsReady() bool

func (*EventEngine) SetKubeClient

func (e *EventEngine) SetKubeClient(client client.Client)

func (*EventEngine) Start

func (e *EventEngine) Start(ctx context.Context) error

EventEngine Processor channels are ready to receive event after the ready channel is closed Error is returned if there is a configuration problem

type EventJsons

type EventJsons []json.RawMessage

type EventProcessorSender

type EventProcessorSender interface {
	Start(ctx context.Context) error
	Process(context.Context, Event) error
	Send(context.Context, UserManifestType) error
}

Must Start, Process and Send

type EventReporter

type EventReporter struct {
	// contains filtered or unexported fields
}

func NewEventReporter

func NewEventReporter(
	log logr.Logger,
	config *Config,
	dataService *dataservice.DataService,
) (*EventReporter, error)

func (*EventReporter) Report

func (r *EventReporter) Report(metadata Metadata, manifestType string, eventJsons EventJsons) error

type Events

type Events []Event

type Manifest

type Manifest struct {
	Type    string `json:"type,omitempty"`
	Version string `json:"version,omitempty"`
}

type Metadata

type Metadata map[string]string

type ProcessorSender

type ProcessorSender struct {
	EventChan chan Event

	EventProcessorSender
	// contains filtered or unexported fields
}

Process events on the Event Channel and send when conditions are met

func (*ProcessorSender) Process

func (p *ProcessorSender) Process(ctx context.Context, event Event) error

func (*ProcessorSender) ReportOne

func (p *ProcessorSender) ReportOne(event Event) error

Immediately send a single event in a single report to DataService, bypassing the accumulator

func (*ProcessorSender) Send

func (p *ProcessorSender) Send(ctx context.Context, userManifestType UserManifestType) error

Send eventJsons for [user][manifestType]

func (*ProcessorSender) SendAll

func (p *ProcessorSender) SendAll(ctx context.Context) error

Send all eventJsons for every [user][manifestType]

func (*ProcessorSender) Start

func (p *ProcessorSender) Start(ctx context.Context, ready chan bool) error

func (*ProcessorSender) UpdateErrorStatus

func (p *ProcessorSender) UpdateErrorStatus(ctx context.Context) error

type ReportData

type ReportData struct {
	Metadata   Metadata          `json:"metadata,omitempty"`
	EventJsons []json.RawMessage `json:"data"`
}

type TarGzipPool

type TarGzipPool struct {
	gp.GzipPool
}

func (*TarGzipPool) TarGzip

func (pool *TarGzipPool) TarGzip(src string, dest string) error

type UserConfigs

type UserConfigs struct {
	// contains filtered or unexported fields
}

func (*UserConfigs) GetMetadata

func (a *UserConfigs) GetMetadata(userName string) Metadata

func (*UserConfigs) HasUser

func (a *UserConfigs) HasUser(userName string) bool

func (*UserConfigs) SetUserConfigs

func (a *UserConfigs) SetUserConfigs(userConfigs []v1alpha1.UserConfig)

type UserManifestType

type UserManifestType struct {
	User         string
	ManifestType string
}