replay

Documentation

Index

• Variables
• type BatchArchiver
• type Clock
• type Config
  • func NewConfig() Config
  • func (c Config) Validate() error
• type DataSource
• type Diagnostic
• type ExecutionStats
• type Recording
  • func (r Recording) MarshalBinary() ([]byte, error)
  • func (r Recording) ObjectID() string
  • func (r *Recording) UnmarshalBinary(data []byte) error
• type RecordingDAO
• type RecordingType
• type Replay
  • func (r Replay) MarshalBinary() ([]byte, error)
  • func (r Replay) ObjectID() string
  • func (r *Replay) UnmarshalBinary(data []byte) error
• type ReplayDAO
• type Service
  • func NewService(conf Config, d Diagnostic) *Service
  • func (s *Service) Close() error
  • func (s *Service) Open() error
• type Status

Variables

var (
	ErrRecordingExists   = errors.New("recording already exists")
	ErrNoRecordingExists = errors.New("no recording exists")

	ErrReplayExists   = errors.New("replay already exists")
	ErrNoReplayExists = errors.New("no replay exists")
)

Types

type BatchArchiver

type BatchArchiver interface {
	io.Closer
	Archive(idx int) (io.Writer, error)
}

type Clock

type Clock int

const (
	Fast Clock = iota
	Real
)

type Config

type Config struct {
	Dir string `toml:"dir"`
}

func NewConfig

func NewConfig() Config

func (Config) Validate

func (c Config) Validate() error

type DataSource

type DataSource interface {
	Size() (int64, error)
	Remove() error
	StreamWriter() (io.WriteCloser, error)
	StreamReader() (io.ReadCloser, error)
	BatchArchiver() (BatchArchiver, error)
	BatchReaders() ([]io.ReadCloser, error)
}

type Diagnostic

type Diagnostic interface {
	Error(msg string, err error, ctx ...keyvalue.T)
	Debug(msg string, ctx ...keyvalue.T)
}

type ExecutionStats

type ExecutionStats struct {
	TaskStats map[string]interface{}
	NodeStats map[string]map[string]interface{}
}

type Recording

type Recording struct {
	ID string
	// URL for stored Recording data. Currently only file:// is supported.
	DataURL  string
	Type     RecordingType
	Size     int64
	Date     time.Time
	Error    string
	Status   Status
	Progress float64
}

func (Recording) MarshalBinary

func (r Recording) MarshalBinary() ([]byte, error)

func (Recording) ObjectID

func (r Recording) ObjectID() string

func (*Recording) UnmarshalBinary

func (r *Recording) UnmarshalBinary(data []byte) error

type RecordingDAO

type RecordingDAO interface {
	// Retrieve a recording
	Get(id string) (Recording, error)

	// Create a recording.
	// ErrRecordingExists is returned if a recording already exists with the same ID.
	Create(recording Recording) error

	// Replace an existing recording.
	// ErrNoRecordingExists is returned if the recording does not exist.
	Replace(recording Recording) error

	// Delete a recording.
	// It is not an error to delete an non-existent recording.
	Delete(id string) error

	// List recordings matching a pattern.
	// The pattern is shell/glob matching see https://golang.org/pkg/path/#Match
	// Offset and limit are pagination bounds. Offset is inclusive starting at index 0.
	// More results may exist while the number of returned items is equal to limit.
	List(pattern string, offset, limit int) ([]Recording, error)

	// Rebuild fixes all indexes of the data.
	Rebuild() error
}

Data access object for Recording data.

type RecordingType

type RecordingType int

const (
	StreamRecording RecordingType = iota
	BatchRecording
)

type Replay

type Replay struct {
	ID            string
	RecordingID   string
	TaskID        string
	RecordingTime bool
	Clock         Clock
	Date          time.Time
	Error         string
	Status        Status
	Progress      float64
	// Stores snapshot of finished replayed Task status
	ExecutionStats ExecutionStats
}

func (Replay) MarshalBinary

func (r Replay) MarshalBinary() ([]byte, error)

func (Replay) ObjectID

func (r Replay) ObjectID() string

func (*Replay) UnmarshalBinary

func (r *Replay) UnmarshalBinary(data []byte) error

type ReplayDAO

type ReplayDAO interface {
	// Retrieve a replay
	Get(id string) (Replay, error)

	// Create a replay.
	// ErrReplayExists is returned if a replay already exists with the same ID.
	Create(replay Replay) error

	// Replace an existing replay.
	// ErrNoReplayExists is returned if the replay does not exist.
	Replace(replay Replay) error

	// Delete a replay.
	// It is not an error to delete an non-existent replay.
	Delete(id string) error

	// List replays matching a pattern.
	// The pattern is shell/glob matching see https://golang.org/pkg/path/#Match
	// Offset and limit are pagination bounds. Offset is inclusive starting at index 0.
	// More results may exist while the number of returned items is equal to limit.
	List(pattern string, offset, limit int) ([]Replay, error)

	// Rebuild rebuilds all indexes for the storage
	Rebuild() error
}

Data access object for Recording data.

type Service

type Service struct {
	StorageService interface {
		Store(namespace string) storage.Interface
		Register(name string, store storage.StoreActioner)
	}
	TaskStore interface {
		Load(id string) (*kapacitor.Task, error)
	}
	HTTPDService interface {
		AddRoutes([]httpd.Route) error
		DelRoutes([]httpd.Route)
	}
	InfluxDBService interface {
		NewNamedClient(name string) (influxdb.Client, error)
	}
	TaskMasterLookup interface {
		Get(string) *kapacitor.TaskMaster
		Set(*kapacitor.TaskMaster)
		Delete(*kapacitor.TaskMaster)
	}
	TaskMaster interface {
		NewFork(name string, dbrps []kapacitor.DBRP, measurements []string) (edge.StatsEdge, error)
		DelFork(name string)
		New(name string) *kapacitor.TaskMaster
		Stream(name string) (kapacitor.StreamCollector, error)
	}
	// contains filtered or unexported fields
}

Handles recording, starting, and waiting on replays

func NewService

func NewService(conf Config, d Diagnostic) *Service

Create a new replay master.

func (*Service) Close

func (s *Service) Close() error

func (*Service) Open

func (s *Service) Open() error

type Status

type Status int

const (
	Failed Status = iota
	Running
	Finished
)