provisioners

Documentation

Index

• Constants
• Variables
• func CreateDispatcherService(ctx context.Context, client runtimeClient.Client, ...) (*corev1.Service, error)
• func CreateK8sService(ctx context.Context, client runtimeClient.Client, c *eventingv1alpha1.Channel) (*corev1.Service, error)
• func CreateVirtualService(ctx context.Context, client runtimeClient.Client, ...) (*istiov1alpha3.VirtualService, error)
• func DispatcherLabels(ccpName string) map[string]string
• func NewLoggingConfig() *logging.Config
• func NewProvisionerLoggerFromConfig(config *logging.Config) *zap.SugaredLogger
• func RemoveFinalizer(o metav1.Object, finalizerName string)
• func UpdateChannel(ctx context.Context, client runtimeClient.Client, u *eventingv1alpha1.Channel) error
• func UpdateClusterChannelProvisionerStatus(ctx context.Context, client runtimeClient.Client, ...) error
• type AddFinalizerResult
  • func AddFinalizer(o metav1.Object, finalizerName string) AddFinalizerResult
• type ChannelReference
  • func ParseChannel(host string) (ChannelReference, error)
  • func (r *ChannelReference) String() string
• type DispatchDefaults
• type Dispatcher
• type Message
  • func (m *Message) AppendToHistory(host string)
  • func (m *Message) History() []string
• type MessageDispatcher
  • func NewMessageDispatcher(logger *zap.SugaredLogger) *MessageDispatcher
  • func (d *MessageDispatcher) DispatchMessage(message *Message, destination, reply string, defaults DispatchDefaults) error
• type MessageReceiver
  • func NewMessageReceiver(receiverFunc func(ChannelReference, *Message) error, logger *zap.SugaredLogger) *MessageReceiver
  • func (r *MessageReceiver) HandleRequest(res http.ResponseWriter, req *http.Request)
  • func (r *MessageReceiver) Start(stopCh <-chan struct{}) error

Constants

const (
	PortName   = "http"
	PortNumber = 80
	// EventingChannelLabel carries the name of knative's label for the channel
	EventingChannelLabel = "eventing.knative.dev/channel"
	// EventingProvisionerLabel carries the name of knative's label for the provisioner
	EventingProvisionerLabel = "eventing.knative.dev/provisioner"

	// OldEventingChannelLabel carries the name of knative's old label for the channel
	OldEventingChannelLabel = "channel"
	// OldEventingProvisionerLabel carries the name of knative's old label for the provisioner
	OldEventingProvisionerLabel = "provisioner"
)

const (
	// MessageHistoryHeader is the header containing all channel hosts traversed by the message
	// This is an experimental header: https://github.com/knative/eventing/issues/638
	MessageHistoryHeader    = "ce-knativehistory"
	MessageHistorySeparator = "; "
)

const (
	MessageReceiverPort = 8080
)

MessageReceiver starts a server to receive new messages for the channel dispatcher. The new message is emitted via the receiver function.

Variables

var ErrUnknownChannel = errors.New("unknown channel")

ErrUnknownChannel is returned when a message is received by a channel dispatcher for a channel that does not exist.

Functions

func CreateDispatcherService

func CreateDispatcherService(ctx context.Context, client runtimeClient.Client, ccp *eventingv1alpha1.ClusterChannelProvisioner) (*corev1.Service, error)

func CreateK8sService

func CreateK8sService(ctx context.Context, client runtimeClient.Client, c *eventingv1alpha1.Channel) (*corev1.Service, error)

func CreateVirtualService

func CreateVirtualService(ctx context.Context, client runtimeClient.Client, channel *eventingv1alpha1.Channel, svc *corev1.Service) (*istiov1alpha3.VirtualService, error)

func DispatcherLabels

func DispatcherLabels(ccpName string) map[string]string

func NewLoggingConfig

func NewLoggingConfig() *logging.Config

NewLoggingConfig creates a static logging configuration appropriate for a provisioner. All logging levels are set to Info.

func NewProvisionerLoggerFromConfig

func NewProvisionerLoggerFromConfig(config *logging.Config) *zap.SugaredLogger

NewProvisionerLoggerFromConfig creates a new zap logger for the provisioner component based on the provided configuration

func RemoveFinalizer

func RemoveFinalizer(o metav1.Object, finalizerName string)

func UpdateChannel

func UpdateChannel(ctx context.Context, client runtimeClient.Client, u *eventingv1alpha1.Channel) error

func UpdateClusterChannelProvisionerStatus

func UpdateClusterChannelProvisionerStatus(ctx context.Context, client runtimeClient.Client, u *eventingv1alpha1.ClusterChannelProvisioner) error

Types

type AddFinalizerResult

type AddFinalizerResult bool

AddFinalizerResult is used indicate whether a finalizer was added or already present.

const (
	FinalizerAlreadyPresent AddFinalizerResult = false
	FinalizerAdded          AddFinalizerResult = true
)

func AddFinalizer

func AddFinalizer(o metav1.Object, finalizerName string) AddFinalizerResult

AddFinalizer adds finalizerName to the Object.

type ChannelReference

type ChannelReference struct {
	Namespace string
	Name      string
}

ChannelReference references a Channel within the cluster by name and namespace.

func ParseChannel

func ParseChannel(host string) (ChannelReference, error)

ParseChannel converts the channel's hostname into a channel reference.

func (*ChannelReference) String

func (r *ChannelReference) String() string

type DispatchDefaults

type DispatchDefaults struct {
	Namespace string
}

DispatchDefaults provides default parameter values used when dispatching a message.

type Dispatcher

type Dispatcher interface {
	// DispatchMessage dispatches a message to a destination over HTTP.
	//
	// The destination and reply are DNS names. For names with a single label,
	// the default namespace is used to expand it into a fully qualified name
	// within the cluster.
	DispatchMessage(message *Message, destination, reply string, defaults DispatchDefaults) error
}

type Message

type Message struct {
	// Headers provide metadata about the message payload. All header keys
	// should be lowercase.
	Headers map[string]string `json:"headers,omitempty"`

	// Payload is the raw binary content of the message. The payload format is
	// often described by the 'content-type' header.
	Payload []byte `json:"payload,omitempty"`
}

Message represents an chunk of data within a channel dispatcher. The message contains both a map of string headers and a binary payload. This struct gets mashaled/unmarshaled in order to preserve and pass Header information to the event subscriber.

A message may represent a CloudEvent.

func (*Message) AppendToHistory

func (m *Message) AppendToHistory(host string)

AppendToHistory append a new host at the end of the list of hosts of the message history

func (*Message) History

func (m *Message) History() []string

History returns the list of hosts where the message has been into

type MessageDispatcher

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

MessageDispatcher dispatches messages to a destination over HTTP.

func NewMessageDispatcher

func NewMessageDispatcher(logger *zap.SugaredLogger) *MessageDispatcher

NewMessageDispatcher creates a new message dispatcher that can dispatch messages to HTTP destinations.

func (*MessageDispatcher) DispatchMessage

func (d *MessageDispatcher) DispatchMessage(message *Message, destination, reply string, defaults DispatchDefaults) error

DispatchMessage dispatches a message to a destination over HTTP.

The destination and reply are DNS names. For names with a single label, the default namespace is used to expand it into a fully qualified name within the cluster.

type MessageReceiver

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

Message receiver receives messages.

func NewMessageReceiver

func NewMessageReceiver(receiverFunc func(ChannelReference, *Message) error, logger *zap.SugaredLogger) *MessageReceiver

NewMessageReceiver creates a message receiver passing new messages to the receiverFunc.

func (*MessageReceiver) HandleRequest

func (r *MessageReceiver) HandleRequest(res http.ResponseWriter, req *http.Request)

HandleRequest is an http.Handler function. The request is converted to a Message and emitted to the receiver func.

The response status codes:

202 - the message was sent to subscribers
404 - the request was for an unknown channel
500 - an error occurred processing the request

func (*MessageReceiver) Start

func (r *MessageReceiver) Start(stopCh <-chan struct{}) error

Start begings to receive messages for the receiver.

Only HTTP POST requests to the root path (/) are accepted. If other paths or methods are needed, use the HandleRequest method directly with another HTTP server.

This method will block until a message is received on the stop channel.