cdc

Documentation

Overview

Package cdc contains a http.Handler which can receive webhook events from a CockroachDB CDC changefeed. Row updates and resolved timestamps are written to staging tables. The resolved timestamps are processed as a logical loop.

Index

• Variables
• func ProvideConveyorConfig(cfg *Config) *conveyor.Config
• func ProvideDLQConfig(cfg *Config) *dlq.Config
• func ProvideSchemaWatchConfig(cfg *Config) *schemawatch.Config
• func ProvideScriptConfig(cfg *Config) *script.Config
• func ProvideSequencerConfig(cfg *Config) *sequencer.Config
• type Config
  • func (c *Config) Bind(f *pflag.FlagSet)
  • func (c *Config) Preflight() error
• type Handler
  • func ProvideHandler(auth types.Authenticator, cfg *Config, conv *conveyor.Conveyors, ...) (*Handler, error)
  • func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type WebhookPayload
  • func NewWebhookPayload(batch *types.MultiBatch) (*WebhookPayload, error)
  • func NewWebhookResolved(ts hlc.Time) (*WebhookPayload, error)
• type WebhookPayloadLine

Variables

var Set = wire.NewSet(
	ProvideHandler,
	ProvideConveyorConfig,
	ProvideDLQConfig,
	ProvideSchemaWatchConfig,
	ProvideScriptConfig,
	ProvideSequencerConfig,
	conveyor.Set,
)

Set is used by Wire.

Functions

func ProvideConveyorConfig

func ProvideConveyorConfig(cfg *Config) *conveyor.Config

ProvideConveyorConfig is called by Wire.

func ProvideDLQConfig

func ProvideDLQConfig(cfg *Config) *dlq.Config

ProvideDLQConfig is called by Wire.

func ProvideSchemaWatchConfig

func ProvideSchemaWatchConfig(cfg *Config) *schemawatch.Config

ProvideSchemaWatchConfig is called by Wire.

func ProvideScriptConfig

func ProvideScriptConfig(cfg *Config) *script.Config

ProvideScriptConfig is called by Wire.

func ProvideSequencerConfig

func ProvideSequencerConfig(cfg *Config) *sequencer.Config

ProvideSequencerConfig is called by Wire.

Types

type Config

type Config struct {
	ConveyorConfig  conveyor.Config
	DLQConfig       dlq.Config
	SequencerConfig sequencer.Config
	SchemaWatch     schemawatch.Config
	ScriptConfig    script.Config
	// Discard all incoming HTTP payloads. This is useful for tuning
	// changefeed throughput without considering Replicator performance.
	Discard bool

	// If non-zero, wait half before and after consuming the payload.
	DiscardDelay time.Duration

	// The maximum amount of data to buffer when reading a single line
	// of ndjson input. This can be increased if the source cluster
	// has large blob values.
	NDJsonBuffer int

	// The maximum amount of time that we want to allow an HTTP handler
	// to run for.
	ResponseTimeout time.Duration
}

Config adds CDC-specific configuration to the core logical loop.

func (*Config) Bind

func (c *Config) Bind(f *pflag.FlagSet)

Bind adds configuration flags to the set.

func (*Config) Preflight

func (c *Config) Preflight() error

Preflight implements logical.Config.

type Handler

type Handler struct {
	Authenticator types.Authenticator // Access checks.
	Config        *Config             // Runtime options.
	Conveyors     *conveyor.Conveyors // Mutation delivery to the target.
	TargetPool    *types.TargetPool   // Access to the target cluster.
}

Handler is an http.Handler for processing webhook requests from a CockroachDB changefeed.

func ProvideHandler

func ProvideHandler(
	auth types.Authenticator, cfg *Config, conv *conveyor.Conveyors, pool *types.TargetPool,
) (*Handler, error)

ProvideHandler is called by Wire.

func (*Handler) ServeHTTP

func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)

type WebhookPayload

type WebhookPayload struct {
	Payload  []WebhookPayloadLine `json:"payload"`
	Length   int                  `json:"length"`
	Resolved string               `json:"resolved"`

	// This field is populated by [NewWebhookPayload]. It is not part
	// of any actual payload, but is useful when synthesizing workloads.
	Range hlc.Range `json:"-"`
}

WebhookPayload describes the envelope structure that we expect to see from a CockroachDB webhook changefeed.

func NewWebhookPayload

func NewWebhookPayload(batch *types.MultiBatch) (*WebhookPayload, error)

NewWebhookPayload constructs a payload for the batch.

func NewWebhookResolved

func NewWebhookResolved(ts hlc.Time) (*WebhookPayload, error)

NewWebhookResolved constructs

type WebhookPayloadLine

type WebhookPayloadLine struct {
	After   json.RawMessage `json:"after"`
	Before  json.RawMessage `json:"before"`
	Key     json.RawMessage `json:"key"`
	Topic   string          `json:"topic"`
	Updated string          `json:"updated"`
}

WebhookPayloadLine is an encoding of a single mutation.