data

Documentation

Index

• Constants
• Variables
• func FormatSentReport(r SentReport) string
• type DiskvSpooler
  • func NewDiskvSpooler(logger *pct.Logger, dataDir, trashDir, hostname string, ...) *DiskvSpooler
  • func (s *DiskvSpooler) CancelFiles()
  • func (s *DiskvSpooler) Files() <-chan string
  • func (s *DiskvSpooler) Purge(now time.Time, limits pc.DataSpoolLimits) (int, map[string][]string)
  • func (s *DiskvSpooler) PurgeChan(c chan time.Time)
  • func (s *DiskvSpooler) Read(file string) ([]byte, error)
  • func (s *DiskvSpooler) Reject(file string) error
  • func (s *DiskvSpooler) Remove(file string) error
  • func (s *DiskvSpooler) Start(sz proto.Serializer) error
  • func (s *DiskvSpooler) Status() map[string]string
  • func (s *DiskvSpooler) Stop() error
  • func (s *DiskvSpooler) Write(service string, data interface{}) error
• type Manager
  • func NewManager(logger *pct.Logger, dataDir, trashDir, hostname string, ...) *Manager
  • func (m *Manager) GetConfig() ([]proto.AgentConfig, []error)
  • func (m *Manager) GetDefaults(uuid string) map[string]interface{}
  • func (m *Manager) Handle(cmd *proto.Cmd) *proto.Reply
  • func (m *Manager) Sender() *Sender
  • func (m *Manager) Spooler() Spooler
  • func (m *Manager) Start() error
  • func (m *Manager) Status() map[string]string
  • func (m *Manager) Stop() error
• type Sender
  • func NewSender(logger *pct.Logger, client pct.WebsocketClient) *Sender
  • func (s *Sender) Start(spool Spooler, tickerChan <-chan time.Time, timeout uint, blackhole bool) error
  • func (s *Sender) Status() map[string]string
  • func (s *Sender) Stop() error
• type SenderStats
  • func NewSenderStats(d time.Duration) *SenderStats
  • func (s *SenderStats) Dump() []SentInfo
  • func (s *SenderStats) Report() SentReport
  • func (s *SenderStats) Sent(info SentInfo)
• type SentInfo
• type SentReport
• type Spooler

Constants

const (
	DEFAULT_DATA_ENCODING      = "gzip"
	DEFAULT_DATA_SEND_INTERVAL = 63
	DEFAULT_DATA_MAX_AGE       = 3600 * 24         // 1d
	DEFAULT_DATA_MAX_SIZE      = 1024 * 1024 * 100 // 100 MiB
	DEFAULT_DATA_MAX_FILES     = 1000
)

const (
	MAX_SEND_ERRORS    = 3
	CONNECT_ERROR_WAIT = 3
)

const (
	WRITE_BUFFER = 100
	CACHE_SIZE   = 1024 * 1024 * 8 // 8M
)

Variables

var (
	BaseReportFormat  = "%d files, %s, %s, %s net util, %s net speed"
	ErrorReportFormat = "%d errors, %d API errors, %d timeouts, %d bad files"
)

var DebugStats = false

var ErrSpoolTimeout = errors.New("Timeout spooling data")

Functions

func FormatSentReport

func FormatSentReport(r SentReport) string

Types

type DiskvSpooler

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

http://godoc.org/github.com/peterbourgon/diskv

func NewDiskvSpooler

func NewDiskvSpooler(logger *pct.Logger, dataDir, trashDir, hostname string, limits pc.DataSpoolLimits) *DiskvSpooler

func (*DiskvSpooler) CancelFiles

func (s *DiskvSpooler) CancelFiles()

func (*DiskvSpooler) Files

func (s *DiskvSpooler) Files() <-chan string

func (*DiskvSpooler) Purge

func (s *DiskvSpooler) Purge(now time.Time, limits pc.DataSpoolLimits) (int, map[string][]string)

func (*DiskvSpooler) PurgeChan

func (s *DiskvSpooler) PurgeChan(c chan time.Time)

func (*DiskvSpooler) Read

func (s *DiskvSpooler) Read(file string) ([]byte, error)

func (*DiskvSpooler) Reject

func (s *DiskvSpooler) Reject(file string) error

func (*DiskvSpooler) Remove

func (s *DiskvSpooler) Remove(file string) error

func (*DiskvSpooler) Start

func (s *DiskvSpooler) Start(sz proto.Serializer) error

func (*DiskvSpooler) Status

func (s *DiskvSpooler) Status() map[string]string

func (*DiskvSpooler) Stop

func (s *DiskvSpooler) Stop() error

func (*DiskvSpooler) Write

func (s *DiskvSpooler) Write(service string, data interface{}) error

type Manager

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

func NewManager

func NewManager(logger *pct.Logger, dataDir, trashDir, hostname string, client pct.WebsocketClient) *Manager

func (*Manager) GetConfig

func (m *Manager) GetConfig() ([]proto.AgentConfig, []error)

func (*Manager) GetDefaults

func (m *Manager) GetDefaults(uuid string) map[string]interface{}

func (*Manager) Handle

func (m *Manager) Handle(cmd *proto.Cmd) *proto.Reply

func (*Manager) Sender

func (m *Manager) Sender() *Sender

func (*Manager) Spooler

func (m *Manager) Spooler() Spooler

func (*Manager) Start

func (m *Manager) Start() error

func (*Manager) Status

func (m *Manager) Status() map[string]string

func (*Manager) Stop

func (m *Manager) Stop() error

type Sender

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

func NewSender

func NewSender(logger *pct.Logger, client pct.WebsocketClient) *Sender

func (*Sender) Start

func (s *Sender) Start(spool Spooler, tickerChan <-chan time.Time, timeout uint, blackhole bool) error

func (*Sender) Status

func (s *Sender) Status() map[string]string

func (*Sender) Stop

func (s *Sender) Stop() error

type SenderStats

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

func NewSenderStats

func NewSenderStats(d time.Duration) *SenderStats

func (*SenderStats) Dump

func (s *SenderStats) Dump() []SentInfo

func (*SenderStats) Report

func (s *SenderStats) Report() SentReport

func (*SenderStats) Sent

func (s *SenderStats) Sent(info SentInfo)

type SentInfo

type SentInfo struct {
	Begin    time.Time
	End      time.Time
	SendTime float64
	Files    uint
	Bytes    uint64
	Errs     uint
	ApiErrs  uint
	Timeouts uint
	BadFiles uint
}

type SentReport

type SentReport struct {
	Begin       time.Time
	End         time.Time
	Bytes       string // humanized bytes, e.g. 443.59 kB
	Duration    string // End - Begin, humanized
	Utilization string // bytes / (End - Begin), Mbps
	Throughput  string // bytes / sendTime, Mbps
	Files       uint
	Errs        uint
	ApiErrs     uint
	Timeouts    uint
	BadFiles    uint
}

type Spooler

type Spooler interface {
	Start(proto.Serializer) error
	Stop() error
	Status() map[string]string
	Write(service string, data interface{}) error
	Files() <-chan string
	CancelFiles()
	Read(file string) ([]byte, error)
	Remove(file string) error
	Reject(file string) error
}