operations

README

Pub/Sub Operations

Operations holds helper functions for creating and understanding Pub/Sub related jobs that operate on Cloud Pub/Sub.

Usage

Directly

There are two Jobs that can be made:

• NewSubscriptionOps - Operations related to subscriptions.
• NewTopicOps - Operations related to topics.

Each Job accepts an Action, supported Actions are:

• create, will attempt to create the Pub/Sub Resource.
• delete, will attempt to delete the Pub/Sub Resource.

Within Reconcilers

These jobs are used with PubSubBase and provides the following methods:

• EnsureSubscription - Confirm or Create Subscription.
• EnsureSubscriptionDeleted - Delete Subscription.
• EnsureTopic - Confirm or Create Topic.
• EnsureTopicDeleted - Delete Topic.

Why Jobs?

The Jobs are designed to be run in the namespace in which the PullSubscription exists with the auth provided. The Job has to run local to the resource to avoid copying service accounts or secrets into the cloud-run-events namespace.

The controller will re-reconcile the resource if the create job is deleted, this could be used as a healing operation by the operator if the Cloud Pub/Sub subscription is deleted using gcloud or the Cloud Console by mistake.

Documentation

Index

• func NewSubscriptionOps(arg SubArgs) *batchv1.Job
• func NewTopicOps(arg TopicArgs) *batchv1.Job
• func SubscriptionJobLabels(owner kmeta.OwnerRefable, action string) map[string]string
• func SubscriptionJobName(owner kmeta.OwnerRefable, action string) string
• func TopicJobLabels(owner kmeta.OwnerRefable, action string) map[string]string
• func TopicJobName(owner kmeta.OwnerRefable, action string) string
• type PubSubOps
  • func (s *PubSubOps) CreateClient(ctx context.Context) error
• type SubActionResult
• type SubArgs
• type SubscriptionOps
  • func (s *SubscriptionOps) Run(ctx context.Context) error
• type TopicActionResult
• type TopicArgs
• type TopicOps
  • func (t *TopicOps) Run(ctx context.Context) error

Functions

func NewSubscriptionOps

func NewSubscriptionOps(arg SubArgs) *batchv1.Job

NewSubscriptionOps returns a new batch Job resource.

func NewTopicOps

func NewTopicOps(arg TopicArgs) *batchv1.Job

func SubscriptionJobLabels

func SubscriptionJobLabels(owner kmeta.OwnerRefable, action string) map[string]string

func SubscriptionJobName

func SubscriptionJobName(owner kmeta.OwnerRefable, action string) string

SubscriptionJobName creates the name of a subscription ops job.

func TopicJobLabels

func TopicJobLabels(owner kmeta.OwnerRefable, action string) map[string]string

TopicJobLabels creates a label to find a job again. keys is recommended to be (name, kind, action)

func TopicJobName

func TopicJobName(owner kmeta.OwnerRefable, action string) string

TopicJobName creates the name of a topic ops job.

Types

type PubSubOps

type PubSubOps struct {
	// Environment variable containing project id.
	Project string `envconfig:"PROJECT_ID"`

	// Client is the Pub/Sub client used by Ops.
	Client pubsub.Client
}

func (*PubSubOps) CreateClient

func (s *PubSubOps) CreateClient(ctx context.Context) error

type SubActionResult

type SubActionResult struct {
	// Project is the project id that we used (this might have
	// been defaulted, so we'll expose it so that controller can
	// reflect this in the Status).
	ProjectId string `json:"projectId,omitempty"`

	// Reason is the reason of the result.
	Reason string `json:"reason,omitempty"`
}

TODO: This is currently only used on success to communicate the project status. If there's something else that could be useful to communicate to the controller, add them here.

type SubArgs

type SubArgs struct {
	// UID is the UID of the resource that caused this action to be
	// taken. It will be added to the pod template as a label as
	// "resource-uid"
	UID string

	Image               string
	Action              string
	ProjectID           string
	TopicID             string
	SubscriptionID      string
	AckDeadline         time.Duration
	RetainAckedMessages bool
	RetentionDuration   time.Duration
	Secret              corev1.SecretKeySelector
	Owner               kmeta.OwnerRefable
}

SubArgs are the configuration required to make a NewSubscriptionOps.

type SubscriptionOps

type SubscriptionOps struct {
	PubSubOps

	// Action is the operation the job should run.
	// Options: [exists, create, delete]
	Action string `envconfig:"ACTION" required:"true"`

	// Topic is the environment variable containing the PubSub Topic being
	// subscribed to's name. In the form that is unique within the project.
	// E.g. 'laconia', not 'projects/my-gcp-project/topics/laconia'.
	Topic string `envconfig:"PUBSUB_TOPIC_ID" required:"true"`

	// Subscription is the environment variable containing the name of the
	// subscription to use.
	Subscription string `envconfig:"PUBSUB_SUBSCRIPTION_ID" required:"true"`

	// AckDeadline is the default maximum time after a subscriber receives a
	// message before the subscriber should acknowledge the message. Defaults
	// to 30 seconds.
	AckDeadline time.Duration `envconfig:"PUBSUB_SUBSCRIPTION_CONFIG_ACK_DEAD" required:"true" default:"30s"`

	// RetainAckedMessages defines whether to retain acknowledged messages. If
	// true, acknowledged messages will not be expunged until they fall out of
	// the RetentionDuration window.
	RetainAckedMessages bool `envconfig:"PUBSUB_SUBSCRIPTION_CONFIG_RET_ACKED" required:"true" default:"false"`

	// RetentionDuration defines how long to retain messages in backlog, from
	// the time of publish. If RetainAckedMessages is true, this duration
	// affects the retention of acknowledged messages, otherwise only
	// unacknowledged messages are retained. Defaults to 7 days. Cannot be
	// longer than 7 days or shorter than 10 minutes.
	RetentionDuration time.Duration `envconfig:"PUBSUB_SUBSCRIPTION_CONFIG_RET_DUR" required:"true" default:"168h"`
}

SubscriptionOps defines the configuration to use for this operation.

func (*SubscriptionOps) Run

func (s *SubscriptionOps) Run(ctx context.Context) error

Run will perform the action configured upon a subscription.

type TopicActionResult

type TopicActionResult struct {
	// Project is the project id that we used (this might have
	// been defaulted, so we'll expose it so that controller can
	// reflect this in the Status).
	ProjectId string `json:"projectId,omitempty"`

	// Reason is the reason of the result.
	Reason string `json:"reason,omitempty"`
}

TODO: This is currently only used on success to communicate the project status. If there's something else that could be useful to communicate to the controller, add them here.

type TopicArgs

type TopicArgs struct {
	UID       string
	Image     string
	Action    string
	ProjectID string
	TopicID   string
	Secret    corev1.SecretKeySelector
	Owner     kmeta.OwnerRefable
}

type TopicOps

type TopicOps struct {
	// Environment variable containing project id.
	Project string `envconfig:"PROJECT_ID"`

	// Action is the operation the job should run.
	// Options: [exists, create, delete]
	Action string `envconfig:"ACTION" required:"true"`

	// Topic is the environment variable containing the PubSub Topic being
	// created. In the form that is unique within the project.
	// E.g. 'laconia', not 'projects/my-gcp-project/topics/laconia'.
	Topic string `envconfig:"PUBSUB_TOPIC_ID" required:"true"`
}

func (*TopicOps) Run

func (t *TopicOps) Run(ctx context.Context) error