probe

Documentation

Index

• Variables
• func NewCeForwardClient(opts ForwardClientOptions) (handlers.CeForwardClient, error)
• func NewCePubSubClient(ctx context.Context, pc *pubsub.Client) (handlers.CePubSubClient, error)
• func NewCeReceiverClient(ctx context.Context, livenessChecker *utils.LivenessChecker, ...) (handlers.CeReceiveClient, error)
• func NewK8sClient(ctx context.Context) (c kubernetes.Interface, err error)
• func NewPubSubClient(ctx context.Context, projectID clients.ProjectID) (c *pubsub.Client, err error)
• func NewStorageClient(ctx context.Context) (c *storage.Client, err error)
• type EnvConfig
• type ForwardClientOptions
  • func NewCeForwardClientOptions(port ForwardPort) ForwardClientOptions
  • func NewTestCeForwardClientOptions(listener ForwardListener) ForwardClientOptions
• type ForwardListener
• type ForwardPort
• type Helper
  • func InitializeTestProbeHelper(ctx context.Context, brokerCellBaseUrl string, projectID clients.ProjectID, ...) (*Helper, error)
  • func NewHelper(env EnvConfig, handler handlers.Interface, ...) *Helper
  • func (ph *Helper) CheckLastEventTimes() func(ctx context.Context) error
  • func (ph *Helper) Run(ctx context.Context)
• type ReceiveClientOptions
  • func NewCeReceiverClientOptions(port ReceivePort) ReceiveClientOptions
  • func NewTestCeReceiverClientOptions(listener ReceiveListener) ReceiveClientOptions
• type ReceiveListener
• type ReceivePort

Variables

var HelperSet wire.ProviderSet = wire.NewSet(
	NewHelper,
	NewPubSubClient,
	NewCePubSubClient,
	NewK8sClient,
	NewStorageClient,
	NewCeForwardClient,
	NewCeReceiverClient,
	NewCeReceiverClientOptions,
	NewCeForwardClientOptions,
)

var TestHelperSet wire.ProviderSet = wire.NewSet(
	NewHelper,
	NewCePubSubClient,
	NewCeForwardClient,
	NewCeReceiverClient,
	NewTestCeReceiverClientOptions,
	NewTestCeForwardClientOptions,
)

Functions

func NewCeForwardClient

func NewCeForwardClient(opts ForwardClientOptions) (handlers.CeForwardClient, error)

func NewCePubSubClient

func NewCePubSubClient(ctx context.Context, pc *pubsub.Client) (handlers.CePubSubClient, error)

func NewCeReceiverClient

func NewCeReceiverClient(ctx context.Context, livenessChecker *utils.LivenessChecker, opts ReceiveClientOptions) (handlers.CeReceiveClient, error)

func NewK8sClient

func NewK8sClient(ctx context.Context) (c kubernetes.Interface, err error)

func NewPubSubClient

func NewPubSubClient(ctx context.Context, projectID clients.ProjectID) (c *pubsub.Client, err error)

func NewStorageClient

func NewStorageClient(ctx context.Context) (c *storage.Client, err error)

Types

type EnvConfig

type EnvConfig struct {
	// Environment variable containing the maximum tolerated staleness duration for events processed by the forward and receiver clients
	LivenessStaleDuration time.Duration `envconfig:"LIVENESS_STALE_DURATION" default:"5m"`

	// Environment variable containing the default timeout duration to wait for an event to be delivered, if no custom timeout is specified
	DefaultTimeoutDuration time.Duration `envconfig:"DEFAULT_TIMEOUT_DURATION" default:"2m"`

	// Environment variable containing the maximum timeout duration to wait for an event to be delivered
	MaxTimeoutDuration time.Duration `envconfig:"MAX_TIMEOUT_DURATION" default:"30m"`
}

type ForwardClientOptions

type ForwardClientOptions []cehttp.Option

func NewCeForwardClientOptions

func NewCeForwardClientOptions(port ForwardPort) ForwardClientOptions

func NewTestCeForwardClientOptions

func NewTestCeForwardClientOptions(listener ForwardListener) ForwardClientOptions

type ForwardListener

type ForwardListener net.Listener

type ForwardPort

type ForwardPort int

type Helper

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

Helper is the main probe helper object which contains the metadata and clients shared between all probe Handlers.

func InitializeTestProbeHelper

func InitializeTestProbeHelper(ctx context.Context, brokerCellBaseUrl string, projectID clients.ProjectID, cronStaleDuration time.Duration, helperEnv EnvConfig, forwardListener ForwardListener, receiveListener ReceiveListener, storageClient *storage.Client, psClient *pubsub.Client, k8sClient kubernetes.Interface) (*Helper, error)

func NewHelper

func NewHelper(env EnvConfig, handler handlers.Interface, ceForwardClient handlers.CeForwardClient, ceReceiveClient handlers.CeReceiveClient, livenessCheker *utils.LivenessChecker) *Helper

func (*Helper) CheckLastEventTimes

func (ph *Helper) CheckLastEventTimes() func(ctx context.Context) error

CheckLastEventTimes returns an actionFunc which checks the delay between the current time and last processed event times from the forward and receiver clients. This handler is used by the liveness checker to declare the liveness status of the probe helper.

func (*Helper) Run

func (ph *Helper) Run(ctx context.Context)

Run starts the probe forwarder and receiver. This function should be called after Initialize.

type ReceiveClientOptions

type ReceiveClientOptions []cehttp.Option

func NewCeReceiverClientOptions

func NewCeReceiverClientOptions(port ReceivePort) ReceiveClientOptions

func NewTestCeReceiverClientOptions

func NewTestCeReceiverClientOptions(listener ReceiveListener) ReceiveClientOptions

type ReceiveListener

type ReceiveListener net.Listener

type ReceivePort

type ReceivePort int