plaklet

README

plaklet

plaklet is a small, single-shot task executor built on kloset. It reads one task description as JSON on stdin, runs it (backup, check, …) against kloset connectors linked into the binary, and streams the result back as JSON on stdout.

It is the execution engine that a driver spawns to actually do work — for example plakar-edge, which forwards tasks from a control plane to a remote network. plaklet itself is control-plane-agnostic: it has no dependency on plakman, only on kloset and a set of built-in connectors.

Protocol

plaklet reads a single ExecPayload object on stdin and writes a stream of ExecReply objects (one JSON object per line) on stdout, ending with a terminal success or failure. The shapes are defined in protocol.go.

// stdin — one ExecPayload
{
  "op": "backup",
  "task_config": { "tags": "nightly", "ignore": "*.tmp" },
  "source": {                       // resolved connector configuration
    "integration": { "name": "fs", "version": "1.1.2" },
    "fields": [ { "key": "location", "val": "fs:///data" } ]
  },
  "target": {
    "integration": { "name": "fs", "version": "1.1.2" },
    "fields": [ { "key": "location", "val": "fs:///backups/store" } ]
  }
}

// stdout — a stream of ExecReply, terminal reply last
{"type":"report","report":{ "type":"backup", "backup": { … } }}
{"type":"success"}

Connector secrets are expected to be already resolved into literal field values before the payload reaches plaklet.

Operations

Op	Needs	Description
backup	source + target	Snapshot a source into a store
check	source (a store)	Verify every snapshot in a store
restore	source (a store) + target (a destination)	Export a snapshot to a destination
sync	source + target (both stores)	Copy snapshots from one store to another

restore and sync select snapshots via task_config: snapshot (a specific ID), latest ("true"), and tags (comma-separated). restore requires exactly one match; sync copies every match the target does not already have.

Remaining operations (prune, maintenance) follow the same pattern and can be added in their own files.

Connectors

The backends are linked in-process (see connectors.go) — the same set the public plakar CLI ships: fs, http, ptar, stdio, tar. Add an integration subpackage import to support more.

Build & run

go build -o plaklet .

plaklet -cache /var/cache/plaklet -quiet < payload.json

Flag	Default	Meaning
-cache	(required)	Cache directory (pebble state cache)
-cpu	GOMAXPROCS-1	Number of CPUs to use
-concurrency	NumCPU	Max backup/scan worker concurrency
-quiet	false	Quiet
-pkg		Reserved; connectors are linked in-process

Documentation

Overview

Package plaklet is a single-shot task executor built on kloset. It reads one ExecPayload as JSON from stdin, runs the requested operation (backup, check, ...) against kloset connectors linked in-process, and streams ExecReply messages as JSON to stdout — a terminal success/failure reply last.

It depends only on kloset and a set of built-in connectors (see connectors.go); it has no dependency on plakman. It runs either as the standalone `plaklet` binary (see cmd/plaklet) or embedded in a driver such as plakar-edge, which invokes Main directly.

Index

• Constants
• func Main(args []string) int
• type BackupContent
• type BackupReport
• type CheckReport
• type ChecksReport
• type Configuration
• type ConfigurationField
• type ExecPayload
• type ExecReply
• type IODir
• type IOScope
• type Integration
• type NetworkSample
• type ReplyType
• type Report
• type ResourceSample
• type RestoreReport
• type State
• type StateCounter
• type StoreIO
• type SyncIO
• type SyncReport
• type SyncsReport

Constants

const (
	MaxResourceSamples = 120
	MaxNetworkSamples  = 120
)

MaxResourceSamples / MaxNetworkSamples cap the sample ring buffers; the samplers self-decimate when full so the buffer spans the whole run at a bounded size.

Functions

func Main

func Main(args []string) int

Main runs the plaklet executor with the given argument list (excluding the program name) and returns a process exit code. It is the single entry point for both the standalone binary and an embedding driver like plakar-edge.

Types

type BackupContent

type BackupContent struct {
	Files       uint64 `json:"files"`
	Directories uint64 `json:"directories"`
	Symlinks    uint64 `json:"symlinks"`
	Devices     uint64 `json:"devices"`
	Pipes       uint64 `json:"pipes"`
	Sockets     uint64 `json:"sockets"`
}

type BackupReport

type BackupReport struct {
	SnapshotID           []byte        `json:"snapshot_id"`
	SnapshotCreationTime time.Time     `json:"snapshot_creation_time"`
	Took                 time.Duration `json:"took"`
	Name                 string        `json:"name"`
	Origin               string        `json:"origin"`
	Root                 string        `json:"root"`
	Size                 int           `json:"size"`
	Items                int           `json:"items"`
	Tags                 []string      `json:"tags"`
	Environment          string        `json:"environment"`
	Category             string        `json:"category"`
	Dataset              string        `json:"dataset"`
	DataClasses          []string      `json:"data_classes"`
	LogicalSize          uint64        `json:"logical_size"`
	Content              BackupContent `json:"content"`
	Errors               int           `json:"errors"`
	Store                StoreIO       `json:"store"`
	URN                  string        `json:"urn"`
	URNID                string        `json:"urnid"`
}

type CheckReport

type CheckReport struct {
	SnapshotID []byte        `json:"snapshot_id"`
	Took       time.Duration `json:"took"`
	Store      StoreIO       `json:"store"`
}

type ChecksReport

type ChecksReport struct {
	Took        time.Duration `json:"took"`
	LogicalSize uint64        `json:"logical_size"`
	Errors      uint64        `json:"errors"`
	Checks      []CheckReport `json:"checks"`
}

type Configuration

type Configuration struct {
	Id          string               `json:"id"`
	Revision    int                  `json:"revision"`
	Type        string               `json:"type"`
	Integration Integration          `json:"integration"`
	Name        string               `json:"name"`
	Fields      []ConfigurationField `json:"fields"`
	Environment string               `json:"environment,omitempty"`
	DataClasses []string             `json:"data_classes,omitempty"`
	URN         string               `json:"urn"`
	URNID       uuid.UUID            `json:"urnid"`
}

Configuration is a resolved connector configuration. Fields carry literal values (secrets are resolved by the caller before the payload reaches plaklet), so Provider is expected to be nil.

type ConfigurationField

type ConfigurationField struct {
	Key      string         `json:"key"`
	Provider *Configuration `json:"provider"`
	Val      string         `json:"val"`
}

type ExecPayload

type ExecPayload struct {
	Op         string            `json:"op"`
	TaskConfig map[string]string `json:"task_config"`
	Source     *Configuration    `json:"source"`
	Target     *Configuration    `json:"target"`
}

ExecPayload is the single JSON object plaklet reads from stdin.

type ExecReply

type ExecReply struct {
	Type    ReplyType       `json:"type"`
	Message string          `json:"message,omitempty"`
	Report  json.RawMessage `json:"report,omitempty"`
	State   json.RawMessage `json:"state,omitempty"`
}

ExecReply is one message in plaklet's stdout stream. Report is emitted as raw JSON so the shape is owned here and consumers forward it verbatim.

type IODir

type IODir struct {
	TotalBytes  int64   `json:"total,omitzero"`
	Overall     float64 `json:"overall,omitzero"`
	OverallWall float64 `json:"overall_wall,omitzero"`
}

IODir is one direction of a kloset iostat scope. Overall is active-time throughput; OverallWall is wall-clock throughput (bytes/sec).

type IOScope

type IOScope struct {
	Read  IODir `json:"r,omitzero"`
	Write IODir `json:"w,omitzero"`
}

type Integration

type Integration struct {
	Id      string `json:"id"`
	Name    string `json:"name"`
	Version string `json:"version"`
}

type NetworkSample

type NetworkSample struct {
	At               int64   `json:"at"`
	ReadBytesPerSec  float64 `json:"read_bps,omitzero"`
	WriteBytesPerSec float64 `json:"write_bps,omitzero"`
}

NetworkSample is one read/write throughput reading (bytes/sec). Read is wall-clock, write is active-time. Plot against At (unix millis).

type ReplyType

type ReplyType string

ReplyType enumerates the messages plaklet streams to stdout.

const (
	ReplyInfo    ReplyType = "info"
	ReplyWarning ReplyType = "warning"
	ReplyError   ReplyType = "error"
	ReplyReport  ReplyType = "report"
	ReplyState   ReplyType = "state"
	ReplyFailure ReplyType = "failure"
	ReplySuccess ReplyType = "success"
)

type Report

type Report struct {
	Type    string         `json:"type"`
	Backup  *BackupReport  `json:"backup,omitempty"`
	Check   *ChecksReport  `json:"check,omitempty"`
	Restore *RestoreReport `json:"restore,omitempty"`
	Sync    *SyncsReport   `json:"sync,omitempty"`
}

Report is the top-level object carried by a ReplyReport. Exactly one of the operation-specific fields is set, matching Type.

type ResourceSample

type ResourceSample struct {
	At          int64   `json:"at"`
	CPUPercent  float64 `json:"cpu_percent"`
	MemoryBytes uint64  `json:"memory_bytes"`
}

ResourceSample is one reading of this process's CPU%/RSS. Samples are not uniformly spaced on long runs; plot against At (unix millis).

type RestoreReport

type RestoreReport struct {
	SnapshotID  []byte        `json:"snapshot_id"`
	Took        time.Duration `json:"took"`
	LogicalSize uint64        `json:"logical_size"`
	Content     BackupContent `json:"content"`
	Errors      uint64        `json:"errors"`
	Store       StoreIO       `json:"store"`
}

type State

type State struct {
	SnapshotID string `json:"snapshot_id,omitzero"`
	Phase      string `json:"phase,omitzero"`

	Summary struct {
		Exists      bool   `json:"exists,omitzero"`
		Paths       uint64 `json:"paths,omitzero"`
		Files       uint64 `json:"files,omitzero"`
		Directories uint64 `json:"directories,omitzero"`
		Symlinks    uint64 `json:"symlinks,omitzero"`
		Xattrs      uint64 `json:"xattrs,omitzero"`
		Size        uint64 `json:"size,omitzero"`
	} `json:"summary,omitzero"`

	Paths    StateCounter `json:"paths,omitzero"`
	Dirs     StateCounter `json:"dirs,omitzero"`
	Files    StateCounter `json:"files,omitzero"`
	Xattrs   StateCounter `json:"xattrs,omitzero"`
	Symlinks StateCounter `json:"symlinks,omitzero"`
	Chunks   StateCounter `json:"chunks,omitzero"`
	Objects  StateCounter `json:"objects,omitzero"`

	Result struct {
		Size     uint64  `json:"size,omitzero"`
		Errors   uint64  `json:"errors,omitzero"`
		Duration float64 `json:"duration,omitzero"`
	} `json:"result,omitzero"`

	// CPU/memory of this plaklet process (one process per job).
	Resources []ResourceSample `json:"resources,omitzero"`
	NumCPU    int              `json:"num_cpu,omitzero"`

	// Read/write throughput series, resolved per operation to kloset iostat
	// scopes.
	Network []NetworkSample `json:"network,omitzero"`

	Processed struct {
		Items uint64 `json:"items,omitzero"`
		Bytes uint64 `json:"bytes,omitzero"`
	} `json:"processed,omitzero"`

	// Latest per-scope I/O, keyed by kloset iostat scope name; feeds Network.
	IO map[string]IOScope `json:"io,omitzero"`
}

State mirrors the subset of plakman's reporting.State that plaklet produces: live progress counters, per-scope IO, and the CPU/RAM/throughput sample buffers. It is streamed as the raw JSON of a ReplyState. Keep the JSON tags in lockstep with plakman/reporting.State so the control plane can unmarshal it.

type StateCounter

type StateCounter struct {
	Total      uint64 `json:"total,omitzero"`
	Ok         uint64 `json:"ok,omitzero"`
	Error      uint64 `json:"error,omitzero"`
	Size       uint64 `json:"size,omitzero"`
	Cached     uint64 `json:"cached,omitzero"`
	CachedSize uint64 `json:"cached_size,omitzero"`
}

type StoreIO

type StoreIO struct {
	BytesRead    int64 `json:"bytes_read"`
	BytesWritten int64 `json:"bytes_written"`
}

type SyncIO

type SyncIO struct {
	Origin StoreIO `json:"origin"`
	Target StoreIO `json:"target"`
}

type SyncReport

type SyncReport struct {
	SnapshotID           []byte        `json:"snapshot_id"`
	SnapshotCreationTime time.Time     `json:"snapshot_creation_time"`
	Took                 time.Duration `json:"took"`
	Name                 string        `json:"name"`
	SourceOrigin         string        `json:"source_origin"`
	Root                 string        `json:"root"`
	Size                 int           `json:"size"`
	Items                int           `json:"items"`
	Tags                 []string      `json:"tags"`
	Environment          string        `json:"environment"`
	Category             string        `json:"category"`
	Dataset              string        `json:"dataset"`
	DataClasses          []string      `json:"data_classes"`
	LogicalSize          uint64        `json:"logical_size"`
	Content              BackupContent `json:"content"`
	Origin               StoreIO       `json:"origin"`
	Target               StoreIO       `json:"target"`
}

type SyncsReport

type SyncsReport struct {
	Took        time.Duration `json:"took"`
	LogicalSize uint64        `json:"logical_size"`
	Errors      uint64        `json:"errors"`
	Syncs       []SyncReport  `json:"syncs"`
}