transfer

Documentation

Index

• Variables
• func ExtractAndCacheLayers(tarPath string, cacheDir string, expectedDigests []string) error
• func GetMainManifestItem(tarPath string) (*manifestItem, error)
• func ParseErrorToStatus(err error) string
• func PruneTarball(inPath string, outPath string, allDigests []string, missingDigests []string) error
• func PublishStatus(image, peer, direction, status string)
• func PushImage(targetIP string, port int, fingerprint string, filePath string, ...) error
• func SendControlMessage(targetIP string, port int, fingerprint string, imageName string, author string, ...) error
• func StartReceiver(ctx context.Context, wg *sync.WaitGroup, listener net.Listener, ...)
• func StitchTarball(prunedPath string, reconstructedPath string, layerCacheDir string, ...) error
• type ApprovalRequest
• type DownloadResult
• type Hub
  • func (h *Hub) GetAll() []byte
  • func (h *Hub) Publish(event ProgressEvent)
• type Manager
  • func (m *Manager) CancelApproval(image string) (peer string, ok bool)
  • func (m *Manager) Get(image, peer string) (*progressWriter, bool)
  • func (m *Manager) GetPendingConn(image, peer string) (*PendingTransfer, bool)
  • func (m *Manager) Register(pw *progressWriter)
  • func (m *Manager) RegisterApproval(image, peer string, respChan chan bool)
  • func (m *Manager) RegisterPendingConn(image, peer string, pt *PendingTransfer)
  • func (m *Manager) Unregister(pw *progressWriter)
  • func (m *Manager) UnregisterApproval(image string)
  • func (m *Manager) UnregisterPendingConn(image, peer string)
• type PendingApproval
• type PendingTransfer
• type ProgressEvent
• type StreamHeader
• type ThrottledReader
  • func NewThrottledReader(r io.Reader, bytesPerSec int) *ThrottledReader
  • func (t *ThrottledReader) Read(p []byte) (int, error)
• type TransferRequest
• type TransferResponse

Variables

var GlobalHub = &Hub{
	transfers: make(map[string]*ProgressEvent),
}

var GlobalManager = &Manager{
	transfers:        make(map[string]*progressWriter),
	pendingApprovals: make(map[string]*PendingApproval),
	pendingConns:     make(map[string]*PendingTransfer),
}

Functions

func ExtractAndCacheLayers

func ExtractAndCacheLayers(tarPath string, cacheDir string, expectedDigests []string) error

unzips only the layer.tar files from the full payload and saves them in the cache

func GetMainManifestItem

func GetMainManifestItem(tarPath string) (*manifestItem, error)

returns the manifest item with the most layers from a Docker tarball

func ParseErrorToStatus

func ParseErrorToStatus(err error) string

converts a transfer error into a short status string for the ledger.

func PruneTarball

func PruneTarball(inPath string, outPath string, allDigests []string, missingDigests []string) error

reads a full Docker tarball and streams

func PublishStatus

func PublishStatus(image, peer, direction, status string)

sends a one-off status event (e.g. "waiting", "failed").

func PushImage

func PushImage(targetIP string, port int, fingerprint string, filePath string, imageName string, hash string, author string, imageArch string, tlsConfig *tls.Config, peerName string) error

connects to the remote node, negotiates a delta transfer, and streams the pruned payload

func SendControlMessage

func SendControlMessage(targetIP string, port int, fingerprint string, imageName string, author string, action string, initiator string, tlsConfig *tls.Config) error

opens a new TLS connection to the receiver and sends a control action. author is included so the receiver can find the pull progressWriter .

func StartReceiver

func StartReceiver(ctx context.Context, wg *sync.WaitGroup, listener net.Listener, incomingDir string, approvalChan chan ApprovalRequest, downloadedChan chan DownloadResult, engineLedger *ledger.Ledger, activeTransfers *atomic.Int32, accepting *atomic.Bool)

runs a background TCP server to listen for incoming files

func StitchTarball

func StitchTarball(prunedPath string, reconstructedPath string, layerCacheDir string, allDigests []string, alreadyOwnedDigests []string) error

takes the incoming lightweight tarball, injects missing layers

Types

type ApprovalRequest

type ApprovalRequest struct {
	Req      TransferRequest
	Response chan bool
}

passes the metadata and provides a channel

type DownloadResult

type DownloadResult struct {
	Path      string
	ImageName string
	Peer      string
}

DownloadResult holds the path to the downloaded tarball and the final image name.

type Hub

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

func (*Hub) GetAll

func (h *Hub) GetAll() []byte

GetAll returns all current transfer states as a JSON array. Cleans up completed/failed/rejected transfers that are older than 10 seconds.

func (*Hub) Publish

func (h *Hub) Publish(event ProgressEvent)

type Manager

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

func (*Manager) CancelApproval

func (m *Manager) CancelApproval(image string) (peer string, ok bool)

rejects a pending approval. Returns the peer and if found.

func (*Manager) Get

func (m *Manager) Get(image, peer string) (*progressWriter, bool)

func (*Manager) GetPendingConn

func (m *Manager) GetPendingConn(image, peer string) (*PendingTransfer, bool)

returns the pending transfer if it exists.

func (*Manager) Register

func (m *Manager) Register(pw *progressWriter)

func (*Manager) RegisterApproval

func (m *Manager) RegisterApproval(image, peer string, respChan chan bool)

records the channel while receiveAndApprove blocks.

func (*Manager) RegisterPendingConn

func (m *Manager) RegisterPendingConn(image, peer string, pt *PendingTransfer)

records a cancellable pre-streaming connection.

func (*Manager) Unregister

func (m *Manager) Unregister(pw *progressWriter)

func (*Manager) UnregisterApproval

func (m *Manager) UnregisterApproval(image string)

removes the record after approval is resolved.

func (*Manager) UnregisterPendingConn

func (m *Manager) UnregisterPendingConn(image, peer string)

removes the record.

type PendingApproval

type PendingApproval struct {
	RespChan chan bool
	Peer     string
}

holds the channel for an in-flight receiver approval.

type PendingTransfer

type PendingTransfer struct {
	// CancelConn closes the live data connection, causing negotiate() to fail.
	CancelConn func()
	// SendControl opens a new connection to the receiver and sends a control message.
	SendControl func(action, initiator string)
}

holds callbacks for the pre-streaming phase.

type ProgressEvent

type ProgressEvent struct {
	Direction string  `json:"direction"`
	Image     string  `json:"image"`
	Peer      string  `json:"peer"`
	Progress  float64 `json:"progress"`
	Speed     string  `json:"speed"`
	Status    string  `json:"status"`
	UpdatedAt int64   `json:"updatedAt"`
}

type StreamHeader

type StreamHeader struct {
	PrunedHash string `json:"pruned_hash"`
	PrunedSize int64  `json:"pruned_size"`
}

type ThrottledReader

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

wraps an io.Reader and limits the read rate to bytesPerSec.

func NewThrottledReader

func NewThrottledReader(r io.Reader, bytesPerSec int) *ThrottledReader

creates a new ThrottledReader.

func (*ThrottledReader) Read

func (t *ThrottledReader) Read(p []byte) (int, error)

type TransferRequest

type TransferRequest struct {
	ImageName string   `json:"image"`
	Size      int64    `json:"size"`
	Hash      string   `json:"hash"`
	Author    string   `json:"author"`
	ImageArch string   `json:"image_arch"`
	Layers    []string `json:"layers"`
	IsControl bool     `json:"is_control,omitempty"`
	Action    string   `json:"action,omitempty"`
	Initiator string   `json:"initiator,omitempty"`
}

send JSON metadata before the actual file bytes

type TransferResponse

type TransferResponse struct {
	Approved      bool     `json:"approved"`
	MissingLayers []string `json:"missing_layers"`
}