watch

Documentation

Overview

Package watch defines the CEDAR wire protocol for subscribing to ad changes in a golang collector (or any daemon backed by the classad collections engine).

It is the single source of truth shared by the collector server (which streams events) and the htcondor client (which consumes them). The protocol is a thin framing over ClassAd messages:

client -> server:  command WatchAds, then a request ClassAd
                   [ WatchAdType = "StartdAd"; WatchCursor = "<base64>" ]
server -> client:  a stream of events, each an event-header ClassAd
                   [ WatchKind = <0..4>; WatchKey = "<key>"; WatchCursor = "<base64>" ]
                   immediately followed, for a WatchKind of Upsert, by the ad
                   ClassAd itself (a separate message).

The stream continues until the client disconnects or the server's context is cancelled. Event kinds mirror collections.WatchKind exactly (Upsert=0 .. Resync=4) so a server can forward events without remapping.

Index

• Constants
• func DecodeRequest(ad *classad.ClassAd) (adType, constraint string, cursor []byte, err error)
• func EncodeHeader(kind Kind, key, cursor []byte) *classad.ClassAd
• func EncodeRequest(adType, constraint string, cursor []byte) *classad.ClassAd
• type Kind
  • func DecodeHeader(ad *classad.ClassAd) (kind Kind, key, cursor []byte, err error)
  • func (k Kind) HasAd() bool
  • func (k Kind) String() string

Constants

const (
	WatchCommandBase = 74000

	// WatchAds subscribes to changes for one ad type in the collector.
	WatchAds = WatchCommandBase + 0
)

Command integers for the watch protocol. WatchCommandBase is a Pelican-specific allocation chosen to avoid collision with upstream HTCondor: it reclaims the removed TRANSFERD_BASE region (74000), which sits in a large gap between the grid-manager base (73000) and the credential base (81000) and which upstream has no live command family in. New watch verbs extend upward from the base.

const (
	AttrAdType     = "WatchAdType"     // request: which ad type to watch
	AttrConstraint = "WatchConstraint" // request: optional ClassAd match expression
	AttrKind       = "WatchKind"       // event: the Kind
	AttrKey        = "WatchKey"        // event: the ad's key (Upsert/Delete)
	AttrCursor     = "WatchCursor"     // request: resume token; event: durable cursor
)

Request/event ClassAd attribute names.

Functions

func DecodeRequest

func DecodeRequest(ad *classad.ClassAd) (adType, constraint string, cursor []byte, err error)

DecodeRequest reads a subscribe request ClassAd. A "" constraint means no filter.

func EncodeHeader

func EncodeHeader(kind Kind, key, cursor []byte) *classad.ClassAd

EncodeHeader builds an event-header ClassAd. key is nil for non-keyed events (Reset/Synced/Resync); cursor is nil except for Synced and live events.

func EncodeRequest

func EncodeRequest(adType, constraint string, cursor []byte) *classad.ClassAd

EncodeRequest builds the subscribe request ClassAd. constraint (a ClassAd match expression, e.g. `DAGManJobId == 42`) may be "" to watch all ads of the type; cursor may be nil for a full replay.

Types

type Kind

type Kind int64

Kind is the wire encoding of a watch event kind. The values match collections.WatchKind so the collector can forward them unchanged.

const (
	// KindUpsert carries the full ad for an added or updated key; the ad follows
	// the header as a separate ClassAd message.
	KindUpsert Kind = iota
	// KindDelete signals a key was removed (no ad follows).
	KindDelete
	// KindReset tells the client to discard its state; an authoritative snapshot
	// of Upserts follows, ending at KindSynced.
	KindReset
	// KindSynced marks the end of catch-up; the client is now live. Its cursor is
	// a durable resume point.
	KindSynced
	// KindResync tells the client the live stream fell behind; it must reconnect
	// with its last persisted cursor.
	KindResync
	// KindGoingAway is a server-injected signal (not a collections.WatchKind) that
	// the server is shutting down or restarting: the client should reconnect --
	// possibly to a different collector -- with its last persisted cursor. Unlike
	// Resync (a fell-behind condition), it is sent proactively before the server
	// closes the stream, so a client can distinguish a graceful restart from a
	// drop and back off / fail over accordingly.
	KindGoingAway
)

func DecodeHeader

func DecodeHeader(ad *classad.ClassAd) (kind Kind, key, cursor []byte, err error)

DecodeHeader reads an event-header ClassAd. A returned key/cursor is nil when the corresponding attribute is absent.

func (Kind) HasAd

func (k Kind) HasAd() bool

HasAd reports whether an event of this kind is followed by an ad message.

func (Kind) String

func (k Kind) String() string