sink

Documentation

Index

• type Args
  • func GetArgs() (Args, error)
• type AuthOverride
• type Clients
  • func ConfigureClients(ctx context.Context, clusterConfig *rest.Config) (Clients, error)
• type DefaultAuthOverride
  • func (r DefaultAuthOverride) OverrideAuthentication(sa string, namespace string, log *zap.SugaredLogger, ...) (discoveryClient discoveryclient.ServerResourcesInterface, ...)
• type MetricsHandler
  • func (cont *MetricsHandler) Intercept(mw MetricsInterceptor) http.HandlerFunc
• type MetricsInterceptor
• type Recorder
  • func NewRecorder() (*Recorder, error)
• type Response
• type Sink
  • func (r Sink) CreateResources(triggerNS, sa string, res []json.RawMessage, triggerName, eventID string, ...) error
  • func (r Sink) ExecuteInterceptors(trInt []*triggersv1.TriggerInterceptor, in *http.Request, event []byte, ...) ([]byte, http.Header, *triggersv1.InterceptorResponse, error)
  • func (r Sink) ExecuteTriggerInterceptors(t triggersv1.Trigger, in *http.Request, event []byte, log *zap.SugaredLogger, ...) ([]byte, http.Header, *triggersv1.InterceptorResponse, error)
  • func (r Sink) HandleEvent(response http.ResponseWriter, request *http.Request)
  • func (r Sink) IsValidPayload(eventHandler http.Handler) http.Handler
  • func (s *Sink) NewMetricsRecorderInterceptor() MetricsInterceptor
• type StatusRecorder
  • func (r *StatusRecorder) WriteHeader(status int)

Types

type Args

type Args struct {
	// ElName is the EventListener name.
	ElName string
	// ElNamespace is the EventListener namespace.
	ElNamespace string
	// Port is the port the Sink should listen on.
	Port string
	// ELReadTimeOut defines the read timeout for EventListener Server
	ELReadTimeOut time.Duration
	// ELWriteTimeOut defines the write timeout for EventListener Server
	ELWriteTimeOut time.Duration
	// ELIdleTimeOut defines the read timeout for EventListener Server
	ELIdleTimeOut time.Duration
	// ELTimeOutHandler defines the timeout for Timeout Handler of EventListener Server
	ELTimeOutHandler time.Duration
	// ElHTTPClientReadTimeOut defines the Read timeout for HTTP Client
	ElHTTPClientReadTimeOut time.Duration
	// ElHTTPClientKeepAlive defines the Keep Alive for HTTP Client
	ElHTTPClientKeepAlive time.Duration
	// ElTLSHandshakeTimeout defines the Handshake timeout for HTTP Client
	ElHTTPClientTLSHandshakeTimeout time.Duration
	// ElResponseHeaderTimeout defines the Response Header timeout for HTTP Client
	ElHTTPClientResponseHeaderTimeout time.Duration
	// ElExpectContinueTimeout defines the Expect timeout for HTTP Client
	ElHTTPClientExpectContinueTimeout time.Duration
	// IsMultiNS determines whether el functions as namespaced or clustered
	IsMultiNS bool
	// Key defines the filename for tls Key.
	Key string
	// Cert defines the filename for tls Cert.
	Cert string
	// PayloadValidation defines whether to validate payload or not
	PayloadValidation bool
	// CloudEventURI refers to the location where cloudevent data need to be send
	CloudEventURI string
}

Args define the arguments for Sink.

func GetArgs

func GetArgs() (Args, error)

GetArgs returns the flagged Args

type AuthOverride

type AuthOverride interface {
	OverrideAuthentication(sa string,
		namespace string,
		log *zap.SugaredLogger,
		defaultDiscoveryClient discoveryclient.ServerResourcesInterface,
		defaultDynamicClient dynamic.Interface) (discoveryClient discoveryclient.ServerResourcesInterface,
		dynamicClient dynamic.Interface,
		err error)
}

AuthOverride is an interface that constructs a discovery client for the ServerResourceInterface and a dynamic client for the Tekton Resources, using the token provide as the bearer token in the REST config used to build those client. The other non-credential related parameters for the REST client used are copied from the in cluster config of the event sink.

type Clients

type Clients struct {
	DiscoveryClient discoveryclient.DiscoveryInterface
	RESTClient      restclient.Interface
	TriggersClient  triggersclientset.Interface
	K8sClient       *kubeclientset.Clientset
	CEClient        cloudevent.CEClient
}

Clients define the set of client dependencies Sink requires.

func ConfigureClients

func ConfigureClients(ctx context.Context, clusterConfig *rest.Config) (Clients, error)

ConfigureClients returns the kubernetes and triggers clientsets

type DefaultAuthOverride

type DefaultAuthOverride struct {
}

func (DefaultAuthOverride) OverrideAuthentication

func (r DefaultAuthOverride) OverrideAuthentication(sa string,
	namespace string,
	log *zap.SugaredLogger,
	defaultDiscoverClient discoveryclient.ServerResourcesInterface,
	defaultDynamicClient dynamic.Interface) (discoveryClient discoveryclient.ServerResourcesInterface,
	dynamicClient dynamic.Interface,
	err error)

type MetricsHandler

type MetricsHandler struct {
	Handler http.Handler
}

MetricsHandlerFunc builds on top of http.Handler, and exposes API to intercept with MetricsInterceptor. This allows building complex long chains without complicated struct manipulation

func (*MetricsHandler) Intercept

func (cont *MetricsHandler) Intercept(mw MetricsInterceptor) http.HandlerFunc

Intercept returns back a continuation that will call the handler func to intercept the continuation call.

type MetricsInterceptor

type MetricsInterceptor func(http.ResponseWriter, *http.Request, http.HandlerFunc)

MetricsInterceptor intercepts an HTTP handler invocation, it is passed both response writer and request which after interception can be passed onto the handler function.

type Recorder

type Recorder struct {
	ReportingPeriod time.Duration
	// contains filtered or unexported fields
}

func NewRecorder

func NewRecorder() (*Recorder, error)

NewRecorder creates a new metrics recorder instance to log the TaskRun related metrics

type Response

type Response struct {
	// EventListener is the name of the eventListener.
	// Deprecated: use EventListenerUID instead.
	EventListener string `json:"eventListener"`
	// Namespace is the namespace that the eventListener is running in.
	// Deprecated: use EventListenerUID instead.
	Namespace string `json:"namespace,omitempty"`
	// EventListenerUID is the UID of the EventListener
	EventListenerUID string `json:"eventListenerUID"`
	// EventID is a uniqueID that gets assigned to each incoming request
	EventID string `json:"eventID,omitempty"`
	// ErrorMessage gives message about Error which occurs during event processing
	ErrorMessage string `json:"errorMessage,omitempty"`
}

Response defines the HTTP body that the Sink responds to events with.

type Sink

type Sink struct {
	KubeClientSet          kubernetes.Interface
	TriggersClient         triggersclientset.Interface
	DiscoveryClient        discoveryclient.ServerResourcesInterface
	DynamicClient          dynamic.Interface
	HTTPClient             *http.Client
	CEClient               cloudevent.CEClient
	EventListenerName      string
	EventListenerNamespace string
	Logger                 *zap.SugaredLogger
	Recorder               *Recorder
	Auth                   AuthOverride
	PayloadValidation      bool
	CloudEventURI          string
	// WGProcessTriggers keeps track of triggers or triggerGroups currently being processed
	// Currently only used in tests to wait for all triggers to finish processing
	WGProcessTriggers *sync.WaitGroup
	EventRecorder     record.EventRecorder

	// listers index properties about resources
	EventListenerLister         listers.EventListenerLister
	TriggerLister               listers.TriggerLister
	TriggerBindingLister        listers.TriggerBindingLister
	ClusterTriggerBindingLister listers.ClusterTriggerBindingLister
	TriggerTemplateLister       listers.TriggerTemplateLister
	ClusterInterceptorLister    listersv1alpha1.ClusterInterceptorLister
	InterceptorLister           listersv1alpha1.InterceptorLister
}

Sink defines the sink resource for processing incoming events for the EventListener.

func (Sink) CreateResources

func (r Sink) CreateResources(triggerNS, sa string, res []json.RawMessage, triggerName, eventID string, log *zap.SugaredLogger) error

func (Sink) ExecuteInterceptors

func (r Sink) ExecuteInterceptors(trInt []*triggersv1.TriggerInterceptor, in *http.Request, event []byte, log *zap.SugaredLogger, eventID string, triggerID string, namespace string, extensions map[string]interface{}) ([]byte, http.Header, *triggersv1.InterceptorResponse, error)

ExecuteInterceptor executes all interceptors for the Trigger and returns back the body, header, and InterceptorResponse to use. When TEP-0022 is fully implemented, this function will only return the InterceptorResponse and error.

func (Sink) ExecuteTriggerInterceptors

func (r Sink) ExecuteTriggerInterceptors(t triggersv1.Trigger, in *http.Request, event []byte, log *zap.SugaredLogger, eventID string, extensions map[string]interface{}) ([]byte, http.Header, *triggersv1.InterceptorResponse, error)

func (Sink) HandleEvent

func (r Sink) HandleEvent(response http.ResponseWriter, request *http.Request)

HandleEvent processes an incoming HTTP event for the event listener.

func (Sink) IsValidPayload

func (r Sink) IsValidPayload(eventHandler http.Handler) http.Handler

func (*Sink) NewMetricsRecorderInterceptor

func (s *Sink) NewMetricsRecorderInterceptor() MetricsInterceptor

type StatusRecorder

type StatusRecorder struct {
	http.ResponseWriter
	Status int
}

func (*StatusRecorder) WriteHeader

func (r *StatusRecorder) WriteHeader(status int)