client

package
v0.0.0-...-f182e34 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 14, 2026 License: Apache-2.0 Imports: 70 Imported by: 0

Documentation

Index

Constants

View Source 
const ObjectMetadataHeaderName = "X-Pelican-Object-Metadata"

ObjectMetadataHeaderName is the HTTP header that carries the rendered Structured Fields dictionary on the upload PUT request. Kept as a public constant so other clients can reuse it.

The origin declares the same header name as origin_serve.ObjectMetadataHeader; the two are deliberately duplicated rather than shared because the client and origin_serve packages do not import each other. Keep the two values identical.

Variables

View Source 
var (
	PelicanError error_codes.PelicanError

	// Indicates the origin responded too slowly after the cache tried to download from it
	CacheTimedOutReadingFromOrigin = errors.New("cache timed out waiting on origin")

	// ErrObjectNotFound is returned when the requested remote object does not exist.
	ErrObjectNotFound = errors.New("remote object not found")
)
View Source 
var (
	ErrServerChecksumMissing = errors.New("no checksum information was returned by server but checksums were required by the client")
)
View Source 
var ErrTooManyRequests = errors.New("too many requests: origin is over its share of the transfer pool")

ErrTooManyRequests is returned by TagScheduler.Submit when the scheduler refuses admission because the tag (typically an upstream origin) is already at its share of the transfer engine's worker pool. It mirrors HTTP 429.

Every rejection wraps this sentinel (see SchedulerRejection), so callers can continue to use errors.Is(err, ErrTooManyRequests) to detect a shed while errors.As(err, &SchedulerRejection{}) recovers the specific reason.

View Source 
var ObjectServersToTry int = 3

Number of caches to attempt to use in any invocation

View Source 
var ReservedObjectMetadataKeys = []string{"path", "size", "etag", "created_at"}

ReservedObjectMetadataKeys lists the keys the origin populates itself; the client refuses to forward them so users get a clear error rather than a silent server-side drop.

View Source 
var SkipAll = errors.New("client: skip all")

SkipAll, when returned from a WalkFunc, tells Walk to end the walk cleanly. Walk itself returns nil in that case. Analogous to fs.SkipAll.

View Source 
var SkipSubtree = errors.New("client: skip subtree")

SkipSubtree, when returned from a WalkFunc invoked on a collection entry (or on an error tuple for a subtree that could not be listed), tells Walk not to descend into that subtree but to continue with the next sibling. Analogous to fs.SkipDir.

Functions

func AcquireToken 

func AcquireToken(destination *url.URL, dirResp server_structs.DirectorResponse, opts config.TokenGenerationOpts) (string, error)

func ByteCountSI 

func ByteCountSI(b int64) string

Convert b bytes to a human-friendly string with SI units

For example, ByteCountSI(2000) returns "2 KB"

func CanDisableProxy 

func CanDisableProxy() bool

Determine whether we are allowed to skip the proxy as a fallback

func ContextWithRequestId 

func ContextWithRequestId(ctx context.Context, id string) context.Context

ContextWithRequestId returns a child context that carries the given request ID. Downstream code can retrieve it with RequestIdFromContext.

func CreateSharingUrl 

func CreateSharingUrl(ctx context.Context, objectUrl *url.URL, isWrite bool) (string, error)

func DoCacheInfo 

func DoCacheInfo(ctx context.Context, destination string, options ...TransferOption) (age int, size int64, err error)

Check the cache information of a remote cache

func DoDelete 

func DoDelete(ctx context.Context, remoteDestination string, recursive bool, options ...TransferOption) (err error)

DoDelete queries the director using the DELETE HTTP method, retrieves the token, and initializes the delete operation.

func DoEvict 

func DoEvict(ctx context.Context, remoteObject string, immediate bool, options ...TransferOption) (message string, err error)

DoEvict evicts cached objects matching the given path (or prefix) from the local cache. Token bootstrapping follows the same logic as DoGet: tokens are discovered from the environment, credential files, or negotiated via OAuth when needed.

The remoteObject argument is a pelican:// or osdf:// URL (or a schemeless namespace path when federation discovery is configured).

When immediate is true the objects are deleted right away; otherwise they are marked for priority eviction (purge-first).

func DoShadowIngest 

func DoShadowIngest(ctx context.Context, sourceFile string, originPrefix string, shadowOriginPrefix string, options ...TransferOption) (int64, string, error)

func GetBehavior 

func GetBehavior(behaviorName string) (packerBehavior, error)

func GetDirectorInfoForPath 

func GetDirectorInfoForPath(ctx context.Context, pUrl *pelican_url.PelicanURL, httpMethod string, token string) (parsedResponse server_structs.DirectorResponse, err error)

Retrieve federation namespace information for a given URL.

This is the public API; it always queries the director's default endpoint. Internal callers that need embedded cache-mode behaviour should use getDirectorInfoForPath instead.

func GetObjectServerHostnames 

func GetObjectServerHostnames(ctx context.Context, testFile string) (urls []string, err error)

func HttpDigestFromChecksum 

func HttpDigestFromChecksum(checksumType ChecksumType) string

func IsRetryable 

func IsRetryable(err error) bool

IsRetryable will return true if the error is retryable

func KnownChecksumTypesAsHttpDigest 

func KnownChecksumTypesAsHttpDigest() (result []string)

List all the checksum types known as HTTP digest strings

func NewTokenGenerator 

func NewTokenGenerator(dest *pelican_url.PelicanURL, dirResp *server_structs.DirectorResponse, operation config.TokenOperation, enableAcquire bool) *tokenGenerator

NewTokenGenerator creates a token generator for the given destination and operation. This is the exported entry point used by cmd/token.go; most internal callers should use the unexported newTokenGenerator.

func ParseDirectorInfo 

func ParseDirectorInfo(dirResp *http.Response) (server_structs.DirectorResponse, error)

Given the Director response, parse the headers and construct the ordered list of object servers.

func ParseRemoteAsPUrl 

func ParseRemoteAsPUrl(ctx context.Context, rp string) (*pelican_url.PelicanURL, error)

Given a remote path, use the client's wisdom to parse it as a Pelican URL, including metadata discovery.

This will handle setting up the URL cache, passing along contexts to discovery, and passing the client context/user agent. Calling this should return a fully populated PelicanURL object, including any metadata that was discovered.

func RefreshExpiringCredentials 

func RefreshExpiringCredentials(ctx context.Context, within time.Duration) (int, error)

RefreshExpiringCredentials proactively refreshes stored OAuth2 tokens in the user's credential wallet that will expire within the given window, using a non-interactive refresh-token grant.

For each credential the issuer's token endpoint is resolved either directly (when the credential is keyed by an issuer URL) or via a Director lookup for the namespace prefix. The wallet is read once and, if any token is refreshed, written back; the number of tokens refreshed is returned.

This is intended for long-running daemons (the client agent) that keep the wallet open. Callers should serialize concurrent invocations. Note that the per-transfer client path performs its own lazy refresh of *expired* tokens; keeping this window comfortably larger than typical transfer durations avoids the two paths racing on the same credential.

func RequestIdFromContext 

func RequestIdFromContext(ctx context.Context) (string, bool)

RequestIdFromContext extracts a request ID previously stored with ContextWithRequestId. Returns ("", false) if none is set.

func ResetJobAd 

func ResetJobAd()

Reset the memory-cached copy of the HTCondor job ad

The client will search through the process's environment to find a HTCondor "job ad" and cache its contents in memory; the job ad is used to determine the project name and job ID for the transfer headers.

This function is used to reset the job ad and is intended for use in unit tests that need to reset things from outside the cache package.

func ShouldRetry 

func ShouldRetry(err error) bool

func Walk 

func Walk(ctx context.Context, remoteObject string, fn WalkFunc, options ...TransferOption) (err error)

Walk lists remoteObject and hands each entry to fn in walk order. It is the streaming primitive that DoList and WalkSeq are built on. The callback receives (FileInfo, error) where a non-nil error identifies a specific subtree that could not be listed; a nil return from fn after such an error tells Walk to continue with the next sibling, and SkipSubtree is an explicit synonym for the same behavior. SkipAll ends the walk cleanly. Any other non-nil return from fn aborts the walk with that error.

Walk is a lower-level replacement for the historical DoList (which buffers results into a slice and aborts on the first error). Prefer Walk when the result set may be large or when the caller wants to keep going past per-subtree failures. See cmd/object_du.go for an example.

func WalkMany 

func WalkMany(ctx context.Context, roots []string, parallelism int, fn WalkManyFunc, options ...TransferOption) (err error)

WalkMany walks each entry in roots concurrently, capped at parallelism (or param.Client_WorkerCount when parallelism <= 0), and reuses a single TransferEngine across all walks. fn receives (root, info, err) so callers can attribute each entry back to its argument; return semantics match Walk (nil to continue, SkipSubtree to prune a subtree in that root's walk, SkipAll to end that walk cleanly, other error to abort that walk).

A per-root failure does not stop the other walks; WalkMany returns errors.Join of every root-level error observed, or nil if every walk finished cleanly.

func WalkSeq 

func WalkSeq(ctx context.Context, remoteObject string, options ...TransferOption) iter.Seq2[FileInfo, error]

WalkSeq is a range-over-func (iter.Seq2) adapter over Walk. It walks remoteObject the same way Walk does but presents each entry as one loop iteration, e.g.: 

for info, err := range client.WalkSeq(ctx, url, options...) {
    if err != nil {
        // per-subtree failure; keep going or return err to abort
        continue
    }
    // handle info
}

If the caller breaks out of the loop early the walk is aborted immediately and no synthetic terminal error tuple is yielded. If the walk fails at a step that isn't attributable to a specific path (e.g. Director lookup), one final (zero, err) tuple is yielded so the loop body observes the error before the range ends.

func WithDirectorDebug 

func WithDirectorDebug(ctx context.Context) context.Context

WithDirectorDebug returns a derived context that causes director queries to include the X-Pelican-Debug header, requesting decision information.

Types

type ByteRange 

type ByteRange struct {
	Start int64
	End   int64 // -1 means "to end of file"
}

ByteRange specifies a byte range for partial object transfers Start and End are inclusive byte offsets (0-indexed) End of -1 means "to end of file"

type CacheThrottleError 

type CacheThrottleError struct {
	Reason     string
	RetryAfter time.Duration
	// Endpoint is the cache host that shed the request.
	Endpoint string
	// Err is the wrapped, specific retryable PelicanError.
	Err error
	// contains filtered or unexported fields
}

CacheThrottleError is returned when a cache responds with HTTP 429 because its fair scheduler shed the request. It carries the machine-parseable Reason (from the response body) and the RetryAfter hint (from the Retry-After header) so callers and external retriers can distinguish an unresponsive origin from a merely-slow one and honor the advertised backoff. It wraps the specific retryable Pelican error type, so errors.As(err, &error_codes.PelicanError{}) and IsRetryable both work through the chain.

func (*CacheThrottleError) Error 

func (e *CacheThrottleError) Error() string

func (*CacheThrottleError) Is 

func (e *CacheThrottleError) Is(target error) bool

Is reports a match for ErrTooManyRequests so a throttle observed as a remote 429 satisfies the same errors.Is check as a shed performed by this process's own scheduler.

func (*CacheThrottleError) Unwrap 

func (e *CacheThrottleError) Unwrap() error

type ChecksumInfo 

type ChecksumInfo struct {
	Algorithm ChecksumType
	Value     []byte
}

Value of one checksum calculation

type ChecksumMismatchError 

type ChecksumMismatchError struct {
	Info        ChecksumInfo // The checksum that was calculated by the client
	ServerValue []byte       // The checksum value that was calculated by the server
}

Represents a mismatched checksum

func (*ChecksumMismatchError) Error 

func (e *ChecksumMismatchError) Error() string

type ChecksumType 

type ChecksumType int
const (

	// The checksum algorithms supported by the client
	//
	// Note we have a helper function, KnownChecksumTypes, that returns a list
	// of all the elements enumerated below; do not skip integers in this list
	// or that functionality will break.
	//
	AlgMD5     ChecksumType = iota // Checksum is using the MD5 algorithm
	AlgCRC32C                      // Checksum is using the CRC32C algorithm
	AlgCRC32                       // Checksum is using the CRC32 algorithm
	AlgSHA1                        // Checksum is using the SHA-1 algorithm
	AlgUnknown                     // Unknown checksum algorithm.  Always a "trailer" indicating the last known algorithm.

	AlgDefault = AlgCRC32C // Default checksum algorithm is CRC32C if the client doesn't specify one.

)

func ChecksumFromHttpDigest 

func ChecksumFromHttpDigest(httpDigest string) ChecksumType

func KnownChecksumTypes 

func KnownChecksumTypes() (result []ChecksumType)

List all the checksum types known to the client

type ConnectionSetupError 

type ConnectionSetupError struct {
	URL string
	Err error
}

ConnectionSetupError is an error that is returned when a connection to the remote server fails

func (*ConnectionSetupError) Error 

func (e *ConnectionSetupError) Error() string

func (*ConnectionSetupError) Is 

func (e *ConnectionSetupError) Is(target error) bool

func (*ConnectionSetupError) Unwrap 

func (e *ConnectionSetupError) Unwrap() error

type ConstantSizer 

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

func (*ConstantSizer) BytesComplete 

func (cs *ConstantSizer) BytesComplete() int64

func (*ConstantSizer) Size 

func (cs *ConstantSizer) Size() int64

type DirRespCache 

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

DirRespCache caches DirectorResponse values keyed by namespace prefix.

It supports longest-prefix matching: given a path like "/federation/data/subdir/file.txt", it will match an entry stored under the prefix "/federation/data" (but not "/federation/other").

Concurrent cache misses for paths that would map to the same singleflight key are coalesced: only one director query is issued and all waiters receive the same result.

Entries expire after a configurable TTL. The cache is safe for concurrent use.

Entries are scoped to the federation they were learned from -- the discovery endpoint, which is the host named in the user's pelican:// URL. One TransferEngine is shared across every transfer in a process, and namespace paths are not globally unique, so two federations can readily present the same one. Without that scoping the second transfer would be answered with the first federation's object servers and issuers, and would send its credentials to a host the user never named -- the thing the rest of the client is careful not to do (see canApplyTokenHint).

func NewDirRespCache 

func NewDirRespCache(ttl time.Duration) *DirRespCache

NewDirRespCache creates a new prefix-matching cache for director responses. Entries are considered valid for `ttl` after they are stored.

func (*DirRespCache) Invalidate 

func (c *DirRespCache) Invalidate(federation string, flavor DirRespFlavor, prefix string)

Invalidate removes the entry for the given prefix within federation.

func (*DirRespCache) InvalidateAll 

func (c *DirRespCache) InvalidateAll()

InvalidateAll removes all cached entries.

func (*DirRespCache) Len 

func (c *DirRespCache) Len() int

Len returns the number of entries in the cache (including expired ones that haven't been cleaned up yet).

func (*DirRespCache) Lookup 

func (c *DirRespCache) Lookup(federation string, flavor DirRespFlavor, objectPath string) (server_structs.DirectorResponse, bool)

Lookup finds the longest cached prefix within federation that matches `objectPath`. An entry learned from a different federation never matches, however similar the path.

For example, if the cache contains entries for "/a/b" and "/a", a lookup for "/a/b/c/d.txt" will return the entry for "/a/b".

Returns the cached DirectorResponse and true if a valid (non-expired) entry was found, or the zero value and false otherwise.

func (*DirRespCache) LookupOrLoad 

func (c *DirRespCache) LookupOrLoad(ctx context.Context, federation string, flavor DirRespFlavor, objectPath string, loader DirRespLoader) (server_structs.DirectorResponse, error)

LookupOrLoad checks the cache first; on a miss it calls `loader` exactly once per unique objectPath, coalescing concurrent callers via singleflight.

If the context is cancelled while waiting for an in-flight query, the waiter returns ctx.Err() immediately. The underlying query keeps running so that other waiters (with live contexts) still receive the result.

On success the response is automatically stored in the cache under the prefix returned by the loader.

func (*DirRespCache) Store 

func (c *DirRespCache) Store(federation string, flavor DirRespFlavor, prefix string, objectPath string, resp server_structs.DirectorResponse)

Store saves a DirectorResponse under the given prefix within federation. Any previous entry for the same prefix in the same federation is replaced.

federation is the discovery endpoint the response was learned from, which is the host named in the user's pelican:// URL. A response is only ever handed back to a later lookup naming that same federation: namespace paths are not globally unique, and answering a request for one federation with another's object servers and issuers would send its credentials to a host the user never named.

objectPath is the federation object path (e.g. "/test/file.txt") that was used to obtain this response from the director. It is stripped from each ObjectServer URL so the cached entry contains only the server-side base path. Pass "" if no stripping is needed.

type DirRespFlavor 

type DirRespFlavor struct {
	// Verb is the HTTP method the director was asked about: GET for a
	// read, PUT for a write, COPY for a third-party copy destination.
	Verb string
	// CacheMode routes the query through the director's origin endpoint,
	// which answers with origins instead of caches.  Set by callers that
	// are themselves a cache.
	CacheMode bool
	// Query is the normalized query string of the object URL.  Parameters
	// including directread and prefercached steer matchmaking, so a
	// response obtained under one set does not answer for another.
	Query string
	// Credential fingerprints the bearer token the response was obtained
	// with, empty for an unauthenticated query.  A director may answer a
	// credentialed request differently -- that is the whole reason the
	// transfer paths re-ask once they hold a token -- so such a response
	// must not be handed to a caller bearing a different credential, or to
	// one bearing none.  It is a digest, never the token: cache keys reach
	// debug logs.
	Credential string
}

DirRespFlavor identifies which question was put to the director. A response answers exactly one of them and must never be handed to a caller who asked another: the object servers for a read are caches, for a write the origins that accept it, and for a ?directread read the origins again. Answering a writer with a read's response would both address the write to caches that reject it and disclose the write credential to every one of them -- the thing the rest of the client is careful not to do.

func NewDirRespFlavor 

func NewDirRespFlavor(verb string, cacheMode bool, rawQuery string) DirRespFlavor

NewDirRespFlavor builds a flavor from the verb, the cache-embedded-client mode, and the object URL's raw query. The query is normalized so that orderings of the same parameters share one entry.

func (DirRespFlavor) String 

func (f DirRespFlavor) String() string

String renders the flavor for use inside a cache key.

func (DirRespFlavor) WithCredential 

func (f DirRespFlavor) WithCredential(token string) DirRespFlavor

WithCredential returns a copy of the flavor scoped to one bearer token.

type DirRespLoader 

type DirRespLoader func(ctx context.Context) (resp server_structs.DirectorResponse, prefix string, err error)

DirRespLoader is a function that queries the director for a given object path. It returns the DirectorResponse and the namespace prefix that should be used as the cache key.

type FileInfo 

type FileInfo struct {
	Name         string
	Size         int64
	ModTime      time.Time
	IsCollection bool
	ETag         string            `json:"etag,omitempty"`      // HTTP ETag header value
	Checksums    map[string]string `json:"checksums,omitempty"` // Checksum type (HTTP digest name) to hex-encoded value
}

Our own FileInfo structure to hold information about a file NOTE: this was created to provide more flexibility to information on a file. The fs.FileInfo interface was causing some issues like not always returning a Name attribute ALSO NOTE: the fields are exported so they can be marshalled into JSON, it does not work otherwise

func DoList 

func DoList(ctx context.Context, remoteObject string, options ...TransferOption) (fileInfos []FileInfo, err error)

DoList collects every FileInfo returned by Walk into a slice and returns it. It preserves its historical abort-on-first-error behavior: any walk-time error (including a single unreadable subtree) is returned and partial results are discarded. Prefer Walk or WalkSeq when you want to keep going past per-subtree failures.

func DoStat 

func DoStat(ctx context.Context, destination string, options ...TransferOption) (fileInfo *FileInfo, err error)

DoStat returns metadata about a single remote object or collection.

The path is treated as something to be read unless the caller passes WithStatUploadDestination(true), in which case it is treated as the destination of a pending upload: the Director is queried with PUT and destination-role token options apply. Either way the role-specific token options override the generic ones, exactly as they do for a transfer, so a caller's --source-token or --dest-token is honored here too.

No TransferEngine is built: a stat needs only the Director response, a token, and statHttp's own gowebdav client, so callers can pre-flight a path without paying for a worker pool and its goroutines.

type GlobalStats 

type GlobalStats struct {
	WorkerCount        int
	StarvingCap        int
	ActiveCap          int
	TotalPending       int
	TotalTags          int
	TotalAdmits        uint64
	TotalRejects       uint64
	TotalRejectsGlobal uint64 // subset of TotalRejects: rejected because global pending was full
	TotalRejectsPerTag uint64 // subset of TotalRejects: rejected because per-tag pending was full
}

GlobalStats is the pool-wide scheduler snapshot. The Total* counters are monotonic for the scheduler's lifetime (unlike PerTagStats, whose counters reset when an idle tag is evicted).

type HeaderTimeoutError 

type HeaderTimeoutError struct{}

func (*HeaderTimeoutError) Error 

func (e *HeaderTimeoutError) Error() string

func (*HeaderTimeoutError) Is 

func (e *HeaderTimeoutError) Is(target error) bool

type HttpErrResp 

type HttpErrResp struct {
	Code int
	Str  string
	Err  error
}

func (*HttpErrResp) Error 

func (e *HttpErrResp) Error() string

func (*HttpErrResp) Unwrap 

func (e *HttpErrResp) Unwrap() error

type InvalidByteInChunkLengthError 

type InvalidByteInChunkLengthError struct {
	Err error
}

func (*InvalidByteInChunkLengthError) Error 

func (e *InvalidByteInChunkLengthError) Error() string

func (*InvalidByteInChunkLengthError) Is 

func (e *InvalidByteInChunkLengthError) Is(target error) bool

func (*InvalidByteInChunkLengthError) Unwrap 

func (e *InvalidByteInChunkLengthError) Unwrap() error

type NetworkResetError 

type NetworkResetError struct{}

func (*NetworkResetError) Error 

func (e *NetworkResetError) Error() string

type PelicanFS 

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

PelicanFS implements io.FS for the Pelican data federation. It provides a filesystem-like interface to objects stored in the federation.

func NewPelicanFS 

func NewPelicanFS(ctx context.Context, options ...TransferOption) *PelicanFS

NewPelicanFS creates a new filesystem interface to the Pelican federation. The provided context is used for all operations, and the options are applied to all transfers. If urlPrefix is empty or "/", it defaults to "osdf:///".

func NewPelicanFSWithPrefix 

func NewPelicanFSWithPrefix(ctx context.Context, urlPrefix string, options ...TransferOption) *PelicanFS

NewPelicanFSWithPrefix creates a new filesystem interface with a URL prefix. All paths will be relative to this prefix. If prefix is empty or "/", defaults to "osdf:///".

func (*PelicanFS) Engine 

func (pfs *PelicanFS) Engine() *TransferEngine

Engine returns the transfer engine backing this filesystem. Callers that also make one-off client calls (Stat, List) can pass it to the engine-aware variants so those share this filesystem's director-response cache instead of querying the director again.

func (*PelicanFS) Open 

func (pfs *PelicanFS) Open(name string) (fs.File, error)

Open opens the named file for reading and returns a fs.File that also implements io.ReaderAt, io.Seeker, io.Writer (for write mode), and fs.ReadDirFile (for directories).

func (*PelicanFS) OpenFile 

func (pfs *PelicanFS) OpenFile(name string, flag int) (fs.File, error)

OpenFile opens the named file with specified flags. Supported flags: os.O_RDONLY, os.O_WRONLY, os.O_RDWR, os.O_CREATE

type PelicanFile 

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

PelicanFile represents an open file in the Pelican federation. It implements fs.File, io.ReaderAt, io.Seeker, io.Writer, and fs.ReadDirFile.

Thread-safety: Most fields that can change during the file's lifetime are protected by mu. The currentEndpoint field uses atomic operations for lock-free access during range reads. Functions with "Locked" suffix must be called with mu held.

func (*PelicanFile) Close 

func (pf *PelicanFile) Close() error

Close closes the file, rendering it unusable for I/O. It implements fs.File.

func (*PelicanFile) Read 

func (pf *PelicanFile) Read(p []byte) (n int, err error)

Read reads up to len(p) bytes into p.

func (*PelicanFile) ReadAt 

func (pf *PelicanFile) ReadAt(p []byte, off int64) (n int, err error)

ReadAt reads len(p) bytes into p starting at offset off in the file. It implements io.ReaderAt. Note: ReadAt does not affect the file position.

func (*PelicanFile) ReadDir 

func (pf *PelicanFile) ReadDir(n int) ([]fs.DirEntry, error)

ReadDir reads the contents of the directory and returns a slice of DirEntry values. It implements fs.ReadDirFile. Can be called multiple times to paginate through entries.

func (*PelicanFile) Seek 

func (pf *PelicanFile) Seek(offset int64, whence int) (int64, error)

Seek sets the offset for the next Read operation and returns the new offset. It implements io.Seeker.

func (*PelicanFile) Stat 

func (pf *PelicanFile) Stat() (fs.FileInfo, error)

Stat returns the FileInfo structure describing the file. It implements fs.File.

func (*PelicanFile) Write 

func (pf *PelicanFile) Write(p []byte) (n int, err error)

Write writes len(p) bytes from p to the file. It implements io.Writer.

type PerTagStats 

type PerTagStats struct {
	// Pending, Active, Starving are the current counts at snapshot time.
	Pending  int
	Active   int
	Starving int
	// EMA is the exponentially-weighted moving average of Active over
	// the configured EMAWindow. Used as the weight input for the
	// per-tag round-robin dispatch decision.
	EMA float64
	// Admits and Rejects count admissions/rejections for this tag since
	// the tag was first seen — or since it was last evicted: a tag idle
	// long enough is dropped entirely (see evictIdleTags), and starts
	// from zero if it returns. Lifetime totals live in GlobalStats.
	Admits  uint64
	Rejects uint64
}

PerTagStats is a per-origin snapshot of scheduler state, intended for monitoring / debugging. Values are consistent (all taken under one lock) but stale as soon as they're read.

type PermissionDeniedError 

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

PermissionDeniedError is returned when a 403 status code is received. The message is generated based on the token's validity.

func (*PermissionDeniedError) Error 

func (e *PermissionDeniedError) Error() string

type SchedulerConfig 

type SchedulerConfig struct {
	// PerTagStarvingPercent — upper bound (percentage of worker pool) on
	// how many workers a single tag may hold while its transfers have
	// not yet produced a first byte of data. 0 or ≥100 disables.
	PerTagStarvingPercent int
	// PerTagActivePercent — upper bound (percentage of worker pool) on
	// how many workers a single tag may hold in total (starving +
	// actively-transferring). 0 or ≥100 disables.
	PerTagActivePercent int
	// PendingBufferSize — total number of pending transfers the
	// scheduler will queue across all tags before shedding new admits
	// with ErrTooManyRequests. 0 disables the cap (unbounded queue).
	//
	// Note that the cache-level knob of the same name,
	// Cache.Throttle.PendingBufferSize, reads 0 as "do not build a
	// scheduler at all" -- it never reaches this field. The two zeroes
	// mean opposite things because they act at different layers: here
	// there is a scheduler and its queue is unbounded; there, there is no
	// scheduler.
	PendingBufferSize int
	// PerTagPendingSize — number of pending transfers the scheduler
	// will queue for any single tag before rejecting new admits with
	// ErrTooManyRequests. 0 disables the per-tag cap.
	PerTagPendingSize int
	// EMAWindow — time constant of the per-tag active-worker EMA used
	// to weight the round-robin. Shorter = more reactive, longer =
	// smoother. 0 disables EMA updates (all eligible tags get equal
	// weight).
	EMAWindow time.Duration
}

SchedulerConfig configures a TagScheduler. Zero-values disable the corresponding limit.

type SchedulerRejection 

type SchedulerRejection struct {
	Reason ShedReason
	Tag    string
	// contains filtered or unexported fields
}

SchedulerRejection is the error returned when the scheduler sheds a submission. It wraps ErrTooManyRequests (so errors.Is still matches) and carries the categorized reason plus the tag (upstream origin host) so the downstream HTTP layer can build a structured 429.

func (*SchedulerRejection) Error 

func (e *SchedulerRejection) Error() string

func (*SchedulerRejection) Unwrap 

func (e *SchedulerRejection) Unwrap() error

Unwrap returns the typed, retryable Pelican error for this rejection's reason, which in turn wraps ErrTooManyRequests. Going through the typed error rather than straight to the sentinel is what makes a locally-shed transfer retryable under IsRetryable, matching how the same shed is classified when it is observed remotely as a 429 (CacheThrottleError). errors.Is(err, ErrTooManyRequests) still holds, one link further down.

type SchedulerSnapshot 

type SchedulerSnapshot struct {
	Global GlobalStats
	Tags   map[string]PerTagStats
}

SchedulerSnapshot is a full snapshot of scheduler state.

type ServerPriority 

type ServerPriority struct {
	URL      *url.URL
	Priority int
}

type ShedReason 

type ShedReason string

ShedReason categorizes why the scheduler refused admission. The reason is derived at the moment of shedding from the tag's in-flight composition and the global pending buffer state, and is carried out to the cache's HTTP layer so the client can be told whether the upstream origin is unresponsive, merely slow, or the cache as a whole is saturated. 

const (
	// ShedOriginUnresponsive: the tag's held worker slots are dominated by
	// "starving" fetches (accepted the connection but produced no first byte).
	// The origin looks unresponsive.
	ShedOriginUnresponsive ShedReason = "origin_unresponsive"
	// ShedOriginSlow: the tag is at its active-transfer cap — the origin is
	// delivering data but already holds its fair share of the pool.
	ShedOriginSlow ShedReason = "origin_slow"
	// ShedCacheOverloaded: the global pending buffer is full — the cache is
	// saturated across all origins, not just this one.
	ShedCacheOverloaded ShedReason = "cache_overloaded"
)

type Sizer 

type Sizer interface {
	Size() int64
	BytesComplete() int64
}

type SlowTransferError 

type SlowTransferError struct {
	BytesTransferred int64
	BytesPerSecond   int64
	BytesTotal       int64
	Duration         time.Duration
	CacheAge         time.Duration
}

SlowTransferError is an error that is returned when a transfer takes longer than the configured timeout

func (*SlowTransferError) Error 

func (e *SlowTransferError) Error() (errMsg string)

func (*SlowTransferError) Is 

func (e *SlowTransferError) Is(target error) bool

type StatusCodeError 

type StatusCodeError int

StatusCodeError indicates the server returned a non-200 code.

The wrapper is done to provide a Pelican-based error hierarchy in case we ever decide to have a different underlying download package.

func (*StatusCodeError) Error 

func (e *StatusCodeError) Error() string

func (*StatusCodeError) Is 

func (e *StatusCodeError) Is(target error) bool

type StoppedTransferError 

type StoppedTransferError struct {
	BytesTransferred int64
	StoppedTime      time.Duration
	CacheHit         bool
	Upload           bool
}

Error type for when the transfer started to return data then completely stopped

func (*StoppedTransferError) Error 

func (e *StoppedTransferError) Error() (errMsg string)

func (*StoppedTransferError) Is 

func (e *StoppedTransferError) Is(target error) bool

type SyncLevel 

type SyncLevel int

Different types of synchronization for recursize transfers 

const (
	SyncNone  SyncLevel = iota // When synchronizing, always re-transfer, regardless of existence at destination.
	SyncExist                  // Skip synchronization transfer if the destination exists
	SyncSize                   // Skip synchronization transfer if the destination exists and matches the current source size
)

type TagScheduler 

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

TagScheduler admits transfers into a bounded per-tag FIFO and dispatches them to workers with a weighted random draw across tags. It exists to keep one misbehaving origin from monopolising the transfer engine's worker pool.

func NewTagScheduler 

func NewTagScheduler(workerCount int, cfg SchedulerConfig) *TagScheduler

NewTagScheduler builds a scheduler for a pool of `workerCount` workers. Call Start to run its goroutine.

func (*TagScheduler) Snapshot 

func (s *TagScheduler) Snapshot(ctx context.Context) SchedulerSnapshot

Snapshot returns a consistent snapshot of scheduler state suitable for driving monitoring metrics. Runs on the scheduler goroutine so counters, gauges, and per-tag maps agree with one another.

Returns the zero snapshot if the scheduler has already stopped.

func (*TagScheduler) Start 

func (s *TagScheduler) Start(ctx context.Context, egrp *errgroup.Group, out chan<- *clientTransferFile)

Start begins the scheduler goroutine under `egrp`, dispatching admitted transfers on the out channel. Stop() blocks until the goroutine exits. Start may be called at most once per scheduler; a second call panics (programming error, not a runtime condition).

The goroutine never returns an error: a scheduler shutdown is not a failure of the group it runs in.

func (*TagScheduler) Stop 

func (s *TagScheduler) Stop()

Stop signals the scheduler to exit and waits for the goroutine. It is safe to call multiple times, including concurrently, and safe to call on a scheduler that was never started.

func (*TagScheduler) Submit 

func (s *TagScheduler) Submit(ctx context.Context, tag string, file *clientTransferFile) error

Submit asks the scheduler to admit `file`, tagged with `tag`. Returns nil on accept (the transfer will be dispatched when a worker is free and the tag is under its caps), ErrTooManyRequests on rejection.

Submit attaches hooks to file.file so the workers can signal first-byte and completion back to the scheduler.

type TimestampedError 

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

func (*TimestampedError) Error 

func (te *TimestampedError) Error() string

func (*TimestampedError) Unwrap 

func (te *TimestampedError) Unwrap() error

type TokenProvider 

type TokenProvider interface {
	Get() (string, error)
}

TokenProvider returns a token value, refreshing as needed. Implementations must be safe for concurrent use. See tokenGenerator for the standard implementation.

func StaticTokenProvider 

func StaticTokenProvider(token string) TokenProvider

StaticTokenProvider returns a TokenProvider that always yields the given token string. Useful when the caller has a fixed token and does not need refresh logic.

type TransferAttemptError 

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

Transfer attempt error wraps an error with information about the service/proxy used

func (*TransferAttemptError) Error 

func (tae *TransferAttemptError) Error() (errMsg string)

func (*TransferAttemptError) Is 

func (tae *TransferAttemptError) Is(target error) bool

func (*TransferAttemptError) Unwrap 

func (tae *TransferAttemptError) Unwrap() error

type TransferCallbackFunc 

type TransferCallbackFunc = func(path string, downloaded int64, totalSize int64, completed bool)

type TransferClient 

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

A client to the transfer engine.

func (*TransferClient) CacheInfo 

func (tc *TransferClient) CacheInfo(ctx context.Context, remoteUrl *url.URL, options ...TransferOption) (age int, size int64, err error)

cacheInfo retrieves and returns the age and size of the specified object.

func (*TransferClient) Cancel 

func (tc *TransferClient) Cancel()

Cancel a client

When cancelled, all channels and goroutines associated with the client will close/exit immediately.

func (*TransferClient) Close 

func (tc *TransferClient) Close()

Close the transfer client object

Any subsequent job submissions will cause a panic

func (*TransferClient) NewCopyJob 

func (tc *TransferClient) NewCopyJob(ctx context.Context, src *url.URL, dest *url.URL, recursive bool, options ...TransferOption) (tj *TransferJob, err error)

Create a new third-party copy job for the client.

This creates a transfer that uses the HTTP COPY verb to instruct the destination server to pull data directly from the source, without the client acting as an intermediary.

The returned object can be further customized as desired. This function does not "submit" the job for execution.

func (*TransferClient) NewPrestageJob 

func (tc *TransferClient) NewPrestageJob(ctx context.Context, remoteUrl *url.URL, options ...TransferOption) (tj *TransferJob, err error)

Create a new prestage job for the client

The returned object can be further customized as desired. This function does not "submit" the job for execution.

func (*TransferClient) NewTransferJob 

func (tc *TransferClient) NewTransferJob(ctx context.Context, remoteUrl *url.URL, localPath string, upload bool, recursive bool, options ...TransferOption) (tj *TransferJob, err error)

Create a new transfer job for the client

The returned object can be further customized as desired. This function does not "submit" the job for execution.

func (*TransferClient) Results 

func (tc *TransferClient) Results() chan TransferResults

Return a channel containing the results from the client

func (*TransferClient) Shutdown 

func (tc *TransferClient) Shutdown() (results []TransferResults, err error)

Shutdown the transfer client

Closes the client and waits for all jobs to exit cleanly. Returns any results that were pending when Shutdown was called

func (*TransferClient) Submit 

func (tc *TransferClient) Submit(tj *TransferJob) error

Submit the transfer job to the client for processing

type TransferEngine 

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

An object able to process transfer jobs.

func NewTransferEngine 

func NewTransferEngine(ctx context.Context, opts ...TransferEngineOption) (te *TransferEngine, err error)

Returns a new transfer engine object whose lifetime is tied to the provided context. Will launcher worker goroutines to handle the underlying transfers

With no options the engine runs the client's configured number of workers (Client.WorkerCount) and hands transfers straight to them; see WithWorkerCount and WithScheduler to change either behavior.

func NewTransferEngineWithWorkers deprecated 

func NewTransferEngineWithWorkers(ctx context.Context, workerCount int) (te *TransferEngine, err error)

NewTransferEngineWithWorkers creates a transfer engine with an explicit number of transfer workers.

Deprecated: use NewTransferEngine with WithWorkerCount.

func (*TransferEngine) Close 

func (te *TransferEngine) Close()

Closes the TransferEngine. No new work may be submitted. Any ongoing work will continue

func (*TransferEngine) Delete 

func (te *TransferEngine) Delete(ctx context.Context, remoteDestination string, recursive bool, options ...TransferOption) (err error)

Delete is DoDelete for a caller that already holds an engine; the director response is taken from the engine's cache rather than re-queried. Removing many objects from one namespace otherwise spends a director round trip on each.

func (*TransferEngine) List 

func (te *TransferEngine) List(ctx context.Context, remoteObject string, options ...TransferOption) ([]FileInfo, error)

List is DoList for a caller that already holds an engine.

func (*TransferEngine) NewClient 

func (te *TransferEngine) NewClient(options ...TransferOption) (client *TransferClient, err error)

Create a new client to work with an engine

func (*TransferEngine) Shutdown 

func (te *TransferEngine) Shutdown() error

Initiates a shutdown of the transfer engine. Waits until all workers have finished

func (*TransferEngine) Stat 

func (te *TransferEngine) Stat(ctx context.Context, destination string, options ...TransferOption) (fileInfo *FileInfo, err error)

Stat is DoStat for a caller that already holds an engine -- the cache, for one, which stats before every miss it fills. It is the same lookup; the difference is that the engine's director responses are reused rather than re-queried, which is the whole of what an engine has to offer a stat.

func (*TransferEngine) Walk 

func (te *TransferEngine) Walk(ctx context.Context, remoteObject string, fn WalkFunc, options ...TransferOption) (err error)

Walk is the package-level Walk for a caller that already holds an engine. It skips the per-call engine setup and teardown, and the listing reuses the engine's director responses instead of asking again.

type TransferEngineOption 

type TransferEngineOption func(*transferEngineConfig)

TransferEngineOption customizes the engine returned by NewTransferEngine.

Unlike TransferOption (an alias for the option package's untyped interface), engine options are a distinct function type; the two sets are therefore not interchangeable, so a per-transfer option such as WithCallback cannot be handed to the constructor and silently dropped.

func WithScheduler 

func WithScheduler(scheduler *TagScheduler) TransferEngineOption

WithScheduler wires the engine to a TagScheduler that admits transfers into per-tag FIFOs and dispatches them to workers with a weighted random draw. Callers that don't want per-tag fairness (e.g., single-user CLI transfers) should omit this option.

The scheduler must have been created with NewTagScheduler; it is started by NewTransferEngine and stopped when the engine shuts down. A nil scheduler is ignored.

func WithWorkerCount 

func WithWorkerCount(workerCount int) TransferEngineOption

WithWorkerCount sets the number of transfer workers the engine launches, overriding the Client.WorkerCount default. This lets embedded consumers (e.g. the cache) run with more concurrency than a command-line client would. A workerCount <= 0 is an error at construction.

type TransferErrors 

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

A container object for multiple sub-errors representing transfer failures.

func NewTransferErrors 

func NewTransferErrors() *TransferErrors

Create a new transfer error object

func (*TransferErrors) AddError 

func (te *TransferErrors) AddError(err error)

func (*TransferErrors) AddPastError 

func (te *TransferErrors) AddPastError(err error, timestamp time.Time)

func (*TransferErrors) AllErrorsRetryable 

func (te *TransferErrors) AllErrorsRetryable() bool

Returns true if all errors are retryable. If no errors are present, then returns true

func (*TransferErrors) Error 

func (te *TransferErrors) Error() string

func (*TransferErrors) Unwrap 

func (te *TransferErrors) Unwrap() []error

func (*TransferErrors) UserError 

func (te *TransferErrors) UserError() string

Return a more refined, user-friendly error string

type TransferJob 

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

A representation of a "transfer job". The job can be submitted to the client library, resulting in one or more transfers (if recursive is true). We assume the transfer job is potentially queued for a long time and all the transfers generated by this job will use the same namespace and token.

func (*TransferJob) Cancel 

func (tj *TransferJob) Cancel()

Cancel the transfer job

func (*TransferJob) GetLookupStatus 

func (tj *TransferJob) GetLookupStatus() (ok bool, err error)

Returns the status of the transfer job-to-file(s) lookup

ok is true if the lookup has completed.

func (*TransferJob) ID 

func (tj *TransferJob) ID() string

Get the transfer's ID

type TransferMetadata 

type TransferMetadata struct {
	ContentLength int64     // Size of the HTTP response body (range length for 206 Partial Content; full object for 200 OK)
	ObjectSize    int64     // Full object size (-1 if unknown; parsed from Content-Range for 206, same as ContentLength for 200)
	ETag          string    // ETag header from response
	LastModified  time.Time // Last-Modified header from response
	ContentType   string    // Content-Type header from response
	CacheControl  string    // Cache-Control header from response
}

TransferMetadata contains early metadata about a transfer received from the server before the data transfer begins. This allows making decisions (e.g., ETag verification) before committing to the full transfer.

type TransferOption 

type TransferOption = option.Interface

func WithAcquireToken 

func WithAcquireToken(enable bool) TransferOption

Create an option to specify the token acquisition logic

Token acquisition (e.g., using OAuth2 to get a token when one isn't found in the environment) defaults to `true` but can be disabled with this options

func WithByteRange 

func WithByteRange(start, end int64) TransferOption

Create an option to specify a byte range for partial object downloads

The start and end parameters are inclusive byte offsets (0-indexed). Use end=-1 to download from start to the end of the file. Example: WithByteRange(0, 1023) downloads the first 1024 bytes. Example: WithByteRange(1024, -1) downloads from byte 1024 to the end.

func WithCacheEmbeddedClientMode 

func WithCacheEmbeddedClientMode(enabled bool) TransferOption

WithCacheEmbeddedClientMode controls whether the client runs in "cache-embedded" mode. When enabled, the client queries the director's origin endpoint (/api/v1.0/director/origin/…) instead of the default shortcut endpoint. This causes the director to redirect to origins rather than to caches, which is the correct behaviour when the transfer client is itself embedded inside a cache process.

With it enabled, a GET for /test/file.txt is explicitly routed to the origin endpoint so the cache can fetch from the origin. When disabled (enabled=false), the same GET is routed through the director's shortcut middleware, which redirects to a cache — the correct behaviour for a site-local cache, which appears to the federation as a client and fetches from other caches rather than directly from origins.

func WithCaches 

func WithCaches(caches ...*url.URL) TransferOption

Create an option to override the cache list

func WithCallback 

func WithCallback(callback TransferCallbackFunc) TransferOption

Create an option that provides a callback for a TransferClient

The callback is invoked periodically by one of the transfer workers, with inputs of the local path (e.g., source on upload), the current bytes transferred, and the total object size

func WithCollectionsUrl 

func WithCollectionsUrl(url string) TransferOption

Override collections URL to be used by the TransferClient

func WithDepth 

func WithDepth(depth int) TransferOption

Create an option to specify the maximum depth for recursive listing

The depth parameter controls how deep the recursive listing will go. A depth of 0 means only the specified directory, 1 means one level deep, etc. A depth of -1 means unlimited depth.

func WithDestinationAcquireToken 

func WithDestinationAcquireToken(enable bool) TransferOption

WithDestinationAcquireToken controls automatic token acquisition for the destination side of a transfer. For put operations this is equivalent to WithAcquireToken; for get operations it is a no-op.

func WithDestinationToken 

func WithDestinationToken(token string) TransferOption

WithDestinationToken provides a token for the destination server in a third-party-copy transfer. For get operations, this is a no-op; for put operations it behaves identically to WithToken.

func WithDestinationTokenLocation 

func WithDestinationTokenLocation(location string) TransferOption

WithDestinationTokenLocation provides a token file for the destination server in a third-party-copy transfer. For get operations, this is a no-op; for put operations it behaves identically to WithTokenLocation.

func WithDryRun 

func WithDryRun(enable bool) TransferOption

Create an option to enable dry-run mode

When enabled, the transfer will display what would be copied without actually modifying the destination. Useful for verifying paths and sources before performing actual transfers.

func WithFedToken 

func WithFedToken(provider TokenProvider) TransferOption

WithFedToken provides a federation token that the client sends as an access_token query parameter on the URL to the origin. Unlike the user token (which goes in the Authorization header and is forwarded by the director via the authz query parameter), the federation token is NOT sent to the director — it is appended to the URL only after the director redirect, so it arrives at the origin as a query param. This is compatible with both Go-based and XRootD-based origins.

The provider is queried for a fresh token on each transfer attempt, so callers can pass a refreshable TokenProvider (e.g. one backed by PersistentCache.getFedToken) to handle short-lived tokens that may expire during long downloads.

func WithForcePrestageAPI 

func WithForcePrestageAPI(force bool) TransferOption

Create an option to force use of the Pelican prestage API

When enabled for prestage transfers, the client will return an error if the cache does not support the Pelican prestage API instead of falling back to the traditional method. This is useful for testing to ensure the API is actually being used.

func WithInPlace 

func WithInPlace(inPlace bool) TransferOption

Create an option to specify whether to write files in-place or use temporary files

When inPlace is false (default), files are written to temporary names and atomically renamed on success (similar to rsync's default behavior). When true, files are written directly to their final destination (similar to rsync's --inplace option).

func WithLazyStat 

func WithLazyStat(enable bool) TransferOption

WithLazyStat defers the metadata lookup a read-mode OpenFile would otherwise perform before returning the handle. A reader that only asks for byte ranges never needs the object's size: end-of-file arrives as the server's own 416, and the operations that do need a size (Stat, Seek from the end, ReadDir) fetch one on demand.

This trades a fail-fast open for a round trip per open. A caller reading many objects it already knows exist -- a filesystem walking its own block store, say -- wants that trade; one that opens a path a user typed probably does not. PelicanFS.Open ignores this option and always stats, because io/fs requires opening a missing file to fail.

func WithMetadataChannel 

func WithMetadataChannel(ch chan<- TransferMetadata) TransferOption

Create an option to receive early transfer metadata before data transfer begins

When provided, the channel will receive a TransferMetadata struct containing information like ETag, Content-Length, and Last-Modified as soon as the server response headers are received, but before any data is transferred. This allows the caller to make decisions (e.g., verify ETag matches expected) before committing to the full transfer.

The channel is optional (non-blocking send). If the channel is full or nil, the transfer will proceed without waiting. The caller should ensure the channel has buffer capacity of at least 1.

func WithNonInteractive 

func WithNonInteractive(enable bool) TransferOption

WithNonInteractive controls whether token acquisition may fall back to the interactive OAuth2 device-code flow. When enabled (true), acquisition uses only cached, refreshable, or locally-generatable tokens and fails instead of prompting. This is intended for callers without a controlling terminal, such as the client agent.

func WithObjectMetadata 

func WithObjectMetadata(fields map[string]any) TransferOption

WithObjectMetadata attaches a map of uploader-supplied fields to an upload job. When non-empty, the resulting PUT carries an X-Pelican-Object-Metadata header whose value is the RFC 9651 Structured Fields rendering of the map (see object_metadata.go).

Values must be scalars: string, bool, int / int64, or float64. The keys "path", "size", "etag", and "created_at" are reserved by the origin and refused at validation time. The map is propagated to every transferFile produced by a recursive upload.

func WithObjectMetadataBlob 

func WithObjectMetadataBlob(body []byte, contentType string) TransferOption

WithObjectMetadataBlob attaches an opaque metadata blob to an upload. When non-empty, the upload PUT switches from raw to multipart/form-data with two parts: the metadata blob first, the object body second. The origin's multipart middleware reads the blob into memory (capped by Origin.Metadata.MaxMetadataBytes), streams the object part through the existing POSC pipeline, and forwards the blob byte-for-byte to the configured metadata endpoint as the second part of a multipart/related body.

`contentType` is the on-the-wire Content-Type a receiver sees; pass an empty string to default to "application/octet-stream". Reserved field names ("metadata", "object" by default) are configured on the origin and are not visible to callers — the client always sends the metadata part with the origin's configured metadata-part name.

func WithObjectMetadataBlobFile 

func WithObjectMetadataBlobFile(path string) TransferOption

WithObjectMetadataBlobFile is the file-shaped sibling of WithObjectMetadataBlob. The blob is read lazily during NewTransferJob's option-apply pass; an unreadable path surfaces as an error from DoPut before any network I/O. The blob's Content-Type is sniffed from the file extension (".xml" ⇒ application/xml, ".json" ⇒ application/json, …) and may be overridden via WithObjectMetadataContentType.

func WithObjectMetadataContentType 

func WithObjectMetadataContentType(contentType string) TransferOption

WithObjectMetadataContentType overrides the Content-Type used for the metadata blob supplied via WithObjectMetadataBlobFile. No effect on WithObjectMetadataBlob (which takes the content type as a constructor argument). Last-supplied wins if both WithObjectMetadataContentType and WithObjectMetadataBlob / WithObjectMetadataBlobFile are used.

func WithObjectMetadataFile 

func WithObjectMetadataFile(path string) TransferOption

WithObjectMetadataFile is the file-shaped sibling of WithObjectMetadata. The supplied path must point to a JSON object whose top-level keys are scalar (string / number / boolean) values. The file is read and validated lazily inside NewTransferJob's option-apply pass, so a malformed file or unreadable path surfaces as an error from NewTransferJob (and therefore from DoPut) rather than at upload time. The same reserved-key rules and value-type restrictions described on WithObjectMetadata apply.

If both WithObjectMetadata and WithObjectMetadataFile are supplied, later options win — matching the option-application order used by every other With…() helper here.

func WithReader 

func WithReader(reader io.ReadCloser) TransferOption

Create an option to provide an io.ReadCloser for upload source

When provided, upload data will be read from this reader instead of localPath. The reader will be closed on completion or error.

func WithRecursive 

func WithRecursive(recursive bool) TransferOption

Create an option to enable recursive listing

When enabled, the list operation will recursively traverse all subdirectories

func WithRejectCollections 

func WithRejectCollections(reject bool) TransferOption

Create an option to reject collections during download.

This is off by default, and deliberately so. Pelican's own machinery moves bytes through this same client -- the cache submits ordinary non-recursive download jobs for the blocks it is filling, and the HTCondor plugin does the same on behalf of a batch system -- and none of it has any use for the check: it is not interpreting a command line, and it pays for the endpoint probe the check needs. Turning this on by default would put that probe in front of every cache block fetch. The commands that do interpret a command line ask for it explicitly.

When enabled, the client determines whether the remote path is a collection before downloading it, and fails the transfer if it is. This is for callers that mean to fetch a single object: without it, a collection is downloaded as whatever bytes the origin happens to serve for it, which is at best a listing and at worst an empty file.

The check rides along with the endpoint probe every download already performs, so it does not add a round trip. It is fail-closed: if no endpoint will say whether the path is a collection, the transfer fails rather than proceeding. An endpoint that will not describe the path it is about to serve is not one to take a GET on faith from, and proceeding is how the caller ends up with a directory listing saved as though it were their object. The cost of that choice is that a download fails against an object server that answers neither a byte range nor a PROPFIND; see objectCached for the order those are tried in.

func WithRequestChecksums 

func WithRequestChecksums(types []ChecksumType) TransferOption

Create an option to specify the checksums to request for a given transfer

func WithRequestId 

func WithRequestId(id string) TransferOption

WithRequestId sets a caller-supplied request ID that is propagated as the X-Pelican-JobId header on all HTTP requests made by this transfer. When set, it takes precedence over the HTCondor job-ad lookup. This is used by the cache to thread an incoming client's request ID through to the origin.

func WithRequireChecksum 

func WithRequireChecksum() TransferOption

Indicate that checksum verification is required

func WithSkipChecksums 

func WithSkipChecksums() TransferOption

WithSkipChecksums suppresses the digest request a download otherwise makes once the bytes are in hand.

That request is a HEAD carrying Want-Digest, and it is worth making for an object: it is what verifies the transfer. It is not worth making for a response the caller has already decided it will neither store nor verify -- a cache streaming a collection's directory index, say. Servers are free to ignore Want-Digest, and an XRootD origin asked to digest a collection answers nothing at all, so the request sits until the transport's response header timeout expires. The caller waiting on the download waits with it.

func WithSourceAcquireToken 

func WithSourceAcquireToken(enable bool) TransferOption

WithSourceAcquireToken controls automatic token acquisition for the source side of a transfer. For get operations this is equivalent to WithAcquireToken; for put operations it is a no-op.

func WithSourceToken 

func WithSourceToken(token string) TransferOption

Create an option to provide a source token for a third-party-copy transfer

func WithSourceTokenLocation 

func WithSourceTokenLocation(location string) TransferOption

Create an option to provide a source token location for a third-party-copy transfer

func WithSourceTokenProvider 

func WithSourceTokenProvider(provider TokenProvider) TransferOption

WithSourceTokenProvider supplies a dynamic producer for the source token of a third-party-copy transfer. For a get operation it provides the primary token instead.

func WithStatUploadDestination 

func WithStatUploadDestination(enable bool) TransferOption

WithStatUploadDestination tells DoStat that the path being stat'ed is the destination of a pending upload rather than a source to be read. This changes two things: the Director is queried with PUT, so the stat runs against origins that accept writes instead of against caches, and destination-role token options (WithDestinationToken, WithDestinationTokenLocation) apply instead of source-role ones.

Callers pre-flighting an upload destination must set this. A GET-flavored stat would send the caller's write credential to every cache in the Director's response, and caches cannot answer for a namespace that grants writes without reads.

func WithSynchronize 

func WithSynchronize(level SyncLevel) TransferOption

Create an option to specify the object synchronization level

The synchronization level specifies what to do if the destination object already exists.

func WithToken 

func WithToken(token string) TransferOption

Create an option to provide a specific token to the transfer

The contents of the token will be used as part of the HTTP request

func WithTokenLocation 

func WithTokenLocation(location string) TransferOption

Create an option to override the token locating logic

This will force the transfer to use a specific file for the token contents instead of doing any sort of auto-detection

func WithTokenProvider 

func WithTokenProvider(provider TokenProvider) TransferOption

WithTokenProvider supplies a dynamic producer for the destination (or, for non-copy transfers, the primary) token. The provider is queried on every token request, so callers can back it with a refreshable source such as an encrypted credential store.

func WithWriter 

func WithWriter(writer io.WriteCloser) TransferOption

Create an option to provide an io.WriteCloser for download destination

When provided, downloaded data will be written to this writer instead of localPath. The writer will be closed on completion or error.

type TransferResult 

type TransferResult struct {
	Number            int           `json:"attemptNumber"`     // indicates which attempt this is
	TransferFileBytes int64         `json:"transferFileBytes"` // how much each attempt downloaded
	TimeToFirstByte   time.Duration `json:"timeToFirstByte"`   // how long it took to download the first byte
	TransferEndTime   time.Time     `json:"transferEndTime"`   // when the transfer ends
	TransferTime      time.Duration `json:"transferTime"`      // amount of time we were transferring per attempt (in seconds)
	CacheAge          time.Duration `json:"cacheAge"`          // age of the data reported by the cache
	Endpoint          string        `json:"endpoint"`          // which origin did it use
	ServerVersion     string        `json:"serverVersion"`     // version of the server
	Error             error         `json:"error"`             // what error the attempt returned (if any)
}

type TransferResults 

type TransferResults struct {
	JobId uuid.UUID `json:"jobId"` // The job ID this result corresponds to

	Error             error                        `json:"error"`
	TransferredBytes  int64                        `json:"transferredBytes"`
	ETag              string                       `json:"etag,omitempty"`  // ETag from the server response (GET or PUT)
	ServerChecksums   []ChecksumInfo               `json:"serverChecksums"` // Checksums returned by the server
	ClientChecksums   []ChecksumInfo               `json:"clientChecksums"` // Checksums calculated by the client
	TransferStartTime time.Time                    `json:"transferStartTime"`
	Scheme            string                       `json:"scheme"`
	Source            string                       `json:"source"`
	Attempts          []TransferResult             `json:"attempts"`
	DirectorDecision  *server_structs.RedirectInfo `json:"directorDecision,omitempty"`
	// contains filtered or unexported fields
}

Represents the results of a single object transfer, potentially across multiple attempts / retries.

func DoCopy 

func DoCopy(ctx context.Context, sourceFile string, destination string, recursive bool, options ...TransferOption) (transferResults []TransferResults, err error)

Start the transfer, whether read or write back. Primarily used for backwards compatibility

func DoGet 

func DoGet(ctx context.Context, remoteObject string, localDestination string, recursive bool, options ...TransferOption) (transferResults []TransferResults, err error)
Start of transfer for pelican object get, gets information from the target source before doing our HTTP GET request

remoteObject: the source file/directory you would like to upload localDestination: the end location of the upload recursive: a boolean indicating if the source is a directory or not

func DoPrestage 

func DoPrestage(ctx context.Context, prefixUrl string, options ...TransferOption) (transferResults []TransferResults, err error)

Single-shot call to prestage a single prefix

func DoPut 

func DoPut(ctx context.Context, localObject string, remoteDestination string, recursive bool, options ...TransferOption) (transferResults []TransferResults, err error)
Start of transfer for pelican object put, gets information from the target destination before doing our HTTP PUT request

localObject: the source file/directory you would like to upload remoteDestination: the end location of the upload recursive: a boolean indicating if the source is a directory or not

func (TransferResults) ID 

func (tr TransferResults) ID() string

type UnexpectedEOFError 

type UnexpectedEOFError struct {
	Err error
}

func (*UnexpectedEOFError) Error 

func (e *UnexpectedEOFError) Error() string

func (*UnexpectedEOFError) Is 

func (e *UnexpectedEOFError) Is(target error) bool

func (*UnexpectedEOFError) Unwrap 

func (e *UnexpectedEOFError) Unwrap() error

type WalkFunc 

type WalkFunc func(FileInfo, error) error

WalkFunc is the callback signature accepted by Walk. It mirrors fs.WalkDirFunc: a successful entry arrives as (info, nil); an error attributable to a specific path in the walk arrives as (FileInfo{Name: path, IsCollection: ...}, err). Returning a non-nil error unwinds the walk with that error, except for the SkipSubtree and SkipAll sentinels below which alter control flow without failing the walk.

type WalkManyFunc 

type WalkManyFunc func(root string, info FileInfo, err error) error

WalkManyFunc is the callback accepted by WalkMany. It matches WalkFunc but also identifies which of the caller-supplied roots the entry came from, enabling per-root accumulation without threading additional state through. Because WalkMany dispatches concurrently, fn may be invoked from multiple goroutines; implementations must be safe for concurrent use.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.