adapter

Documentation

Index

• Variables
• func ConvertToPush(ctx context.Context, event cloudevents.Event) cloudevents.Event
• type Adapter
  • func (a *Adapter) Start(ctx context.Context) error
• type CodecObserved
  • func (c CodecObserved) LatencyMs() *stats.Float64Measure
  • func (c CodecObserved) MethodName() string
  • func (c CodecObserved) TraceName() string
• type PubSubMessage
• type PushMessage
• type ReportArgs
• type StatsReporter
  • func NewStatsReporter() StatsReporter

Variables

var (
	// LatencyMs measures the latency in milliseconds for the PullSubscription
	// adapter methods for Pub/Sub.
	LatencyMs = stats.Float64(
		"events.cloud.google.com/pubsub/adapter/latency",
		"The latency in milliseconds for the PullSubscription adapter methods for Pub/Sub.",
		"ms")
)

var (
	// LatencyView is an OpenCensus view that shows http transport method latency.
	LatencyView = &view.View{
		Name:        "pubsub/pullsubscriptions/adapter/latency",
		Measure:     LatencyMs,
		Description: "The distribution of latency inside of PullSubscription adapter for Pub/Sub.",

		Aggregation: view.Distribution(10, 100, 1000, 10000, 30000, 60000),
		TagKeys:     observability.LatencyTags(),
	}
)

Functions

func ConvertToPush

func ConvertToPush(ctx context.Context, event cloudevents.Event) cloudevents.Event

ConvertToPush convert an event to a Pub/Sub style Push payload.

Types

type Adapter

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

	// Environment variable containing the sink URI.
	Sink string `envconfig:"SINK_URI" required:"true"`

	// Environment variable containing the transformer URI.
	Transformer string `envconfig:"TRANSFORMER_URI"`

	// 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"`

	// ExtensionsBase64 is a based64 encoded json string of a map of
	// CloudEvents extensions (key-value pairs) override onto the outbound
	// event.
	ExtensionsBase64 string `envconfig:"K_CE_EXTENSIONS" required:"true"`

	// SendMode describes how the adapter sends events.
	// One of [binary, structured, push]. Default: binary
	SendMode converters.ModeType `envconfig:"SEND_MODE" default:"binary" required:"true"`

	// MetricsConfigJson is a json string of metrics.ExporterOptions.
	// This is used to configure the metrics exporter options, the config is
	// stored in a config map inside the controllers namespace and copied here.
	MetricsConfigJson string `envconfig:"K_METRICS_CONFIG" required:"true"`

	// LoggingConfigJson is a json string of logging.Config.
	// This is used to configure the logging config, the config is stored in
	// a config map inside the controllers namespace and copied here.
	LoggingConfigJson string `envconfig:"K_LOGGING_CONFIG" required:"true"`

	// TracingConfigJson is a JSON string of tracing.Config. This is used to configure tracing. The
	// original config is stored in a ConfigMap inside the controller's namespace. Its value is
	// copied here as a JSON string.
	TracingConfigJson string `envconfig:"K_TRACING_CONFIG" required:"true"`

	// Environment variable containing the namespace.
	Namespace string `envconfig:"NAMESPACE" required:"true"`

	// Environment variable containing the name.
	Name string `envconfig:"NAME" required:"true"`

	// Environment variable containing the resource group. E.g., storages.events.cloud.google.com.
	ResourceGroup string `envconfig:"RESOURCE_GROUP" default:"pullsubscriptions.pubsub.cloud.google.com" required:"true"`
	// contains filtered or unexported fields
}

Adapter implements the Pub/Sub adapter to deliver Pub/Sub messages from a pre-existing topic/subscription to a Sink.

func (*Adapter) Start

func (a *Adapter) Start(ctx context.Context) error

Start starts the adapter. Note: Only call once, not thread safe.

type CodecObserved

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

CodecObserved is a wrapper to append version to observed.

func (CodecObserved) LatencyMs

func (c CodecObserved) LatencyMs() *stats.Float64Measure

LatencyMs implements Observable.LatencyMs

func (CodecObserved) MethodName

func (c CodecObserved) MethodName() string

MethodName implements Observable.MethodName

func (CodecObserved) TraceName

func (c CodecObserved) TraceName() string

TraceName implements Observable.TraceName

type PubSubMessage

type PubSubMessage struct {
	// ID identifies this message. This ID is assigned by the server and is
	// populated for Messages obtained from a subscription.
	// This field is read-only.
	ID string `json:"id,omitempty"`

	// Data is the actual data in the message.
	Data interface{} `json:"data,omitempty"`

	// Attributes represents the key-value pairs the current message
	// is labelled with.
	Attributes map[string]string `json:"attributes,omitempty"`

	// The time at which the message was published. This is populated by the
	// server for Messages obtained from a subscription.
	// This field is read-only.
	PublishTime time.Time `json:"publish_time,omitempty"`
}

PubSubMessage matches the inner message format used by Push Subscriptions.

type PushMessage

type PushMessage struct {
	// Subscription is the subscription ID that received this Message.
	Subscription string `json:"subscription"`
	// Message holds the Pub/Sub message contents.
	Message *PubSubMessage `json:"message,omitempty"`
}

PushMessage represents the format Pub/Sub uses to push events.

type ReportArgs

type ReportArgs struct {
	Namespace     string
	EventType     string
	EventSource   string
	Name          string
	ResourceGroup string
}

type StatsReporter

type StatsReporter interface {
	// ReportEventCount captures the event count. It records one per call.
	ReportEventCount(args *ReportArgs, responseCode int) error
}

StatsReporter defines the interface for sending metrics.

func NewStatsReporter

func NewStatsReporter() StatsReporter

NewStatsReporter creates a reporter that collects and reports metrics.