metering

Documentation

Index

• Constants
• func Initialize(transfer *server.Transfer, task *server.TransferOperation)
• func InitializeWithTags(transfer *server.Transfer, task *server.TransferOperation, ...)
• type MeteringAgent
  • func Agent() MeteringAgent
  • func NewStubAgent(lgr log.Logger) MeteringAgent
  • func WithAgent(agent MeteringAgent) MeteringAgent
• type MeteringOpts
  • func NewMeteringOpts(transfer *server.Transfer, task *server.TransferOperation) (*MeteringOpts, error)
  • func NewMeteringOptsWithTags(transfer *server.Transfer, task *server.TransferOperation, ...) (*MeteringOpts, error)
  • func (rc *MeteringOpts) BaseMetricFields(startTS, finishTS time.Time, tags []string) map[string]interface{}
  • func (rc *MeteringOpts) BaseMetricFieldsWithLabels(startTS, finishTS time.Time, tags []string) map[string]interface{}
  • func (rc *MeteringOpts) BaseMetricFieldsWithTags(startTS, finishTS time.Time, tags []string) map[string]interface{}
  • func (rc *MeteringOpts) SourceID(schema MetricSchema) string
• type Metric
• type MetricSchema
• type MetricState
• type RowsMetric
  • func NewRowsMetric(serializationSchemas []MetricSchema, runtimeType string) *RowsMetric
  • func (rm *RowsMetric) Count(items []abstract.ChangeItem, calculateValuesSize bool)
  • func (rm *RowsMetric) CountForBatch(input base.EventBatch)
  • func (rm *RowsMetric) Reset() *RowsMetricState
• type RowsMetricState
  • func (rms *RowsMetricState) Serialize(baseOpts *MeteringOpts) (map[MetricSchema][]string, error)
• type SizeBucket
• type Stats
  • func EmptyStats() Stats
  • func (s *Stats) Add(value uint64)
  • func (s *Stats) AddBatch(count uint64, avgSize uint64)
• type Writer

Constants

const (
	DisabledSerialization = MetricSchema("disabled")
	InputRowsSchema       = MetricSchema("datatransfer.data.input.v1")
	OutputRowsSchema      = MetricSchema("datatransfer.data.output.v1")
	ComputeCPUSchema      = MetricSchema("datatransfer.compute.cpu.v1")
	ComputeRAMSchema      = MetricSchema("datatransfer.compute.ram.v1")
	RuntimeCPUSchema      = MetricSchema("datatransfer.runtime.cpu.v1")
	RuntimeRAMSchema      = MetricSchema("datatransfer.runtime.ram.v1")
	OutputRowsCountSchema = MetricSchema("datatransfer.rows.v1")
	AnalyticsSchema       = MetricSchema("analytics")
)

Functions

func Initialize

func Initialize(transfer *server.Transfer, task *server.TransferOperation)

func InitializeWithTags

func InitializeWithTags(transfer *server.Transfer, task *server.TransferOperation, runtimeTags map[string]interface{})

Types

type MeteringAgent

type MeteringAgent interface {
	// RunPusher starts background metrics pushing process. RunPusher must not be called after Stop
	// Pusher is stopped either when Stop method is called or incoming Context is Done.
	RunPusher(ctx context.Context, interval time.Duration) error
	// Stop is used to stop metrics pusher (if it was run). Stop must not be called concurrently with RunPusher
	Stop() error
	SetOpts(config *MeteringOpts) error
	CountInputRows(items []abstract.ChangeItem)
	CountOutputRows(items []abstract.ChangeItem)
	CountOutputBatch(input base.EventBatch)
}

func Agent

func Agent() MeteringAgent

func NewStubAgent

func NewStubAgent(lgr log.Logger) MeteringAgent

func WithAgent

func WithAgent(agent MeteringAgent) MeteringAgent

type MeteringOpts

type MeteringOpts struct {
	TransferID      string
	TransferType    string
	FolderID        string
	CloudID         string
	SrcType         string
	DstType         string
	DstMdbClusterID string
	OperationID     string
	OperationType   string
	JobIndex        string
	ComputeVMID     string
	YtOperationID   string
	YtJobID         string
	Host            string
	Runtime         abstract.Runtime
	Tags            map[string]interface{}
}

func NewMeteringOpts

func NewMeteringOpts(transfer *server.Transfer, task *server.TransferOperation) (*MeteringOpts, error)

func NewMeteringOptsWithTags

func NewMeteringOptsWithTags(transfer *server.Transfer, task *server.TransferOperation, runtimeTags map[string]interface{}) (*MeteringOpts, error)

func (*MeteringOpts) BaseMetricFields

func (rc *MeteringOpts) BaseMetricFields(startTS, finishTS time.Time, tags []string) map[string]interface{}

func (*MeteringOpts) BaseMetricFieldsWithLabels

func (rc *MeteringOpts) BaseMetricFieldsWithLabels(startTS, finishTS time.Time, tags []string) map[string]interface{}

func (*MeteringOpts) BaseMetricFieldsWithTags

func (rc *MeteringOpts) BaseMetricFieldsWithTags(startTS, finishTS time.Time, tags []string) map[string]interface{}

func (*MeteringOpts) SourceID

func (rc *MeteringOpts) SourceID(schema MetricSchema) string

type Metric

type Metric interface {
	Reset() MetricState
	// contains filtered or unexported methods
}

type MetricSchema

type MetricSchema string

type MetricState

type MetricState interface {
	Serialize(baseOpts *MeteringOpts) (map[MetricSchema][]string, error)
}

type RowsMetric

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

func NewRowsMetric

func NewRowsMetric(serializationSchemas []MetricSchema, runtimeType string) *RowsMetric

func (*RowsMetric) Count

func (rm *RowsMetric) Count(items []abstract.ChangeItem, calculateValuesSize bool)

func (*RowsMetric) CountForBatch

func (rm *RowsMetric) CountForBatch(input base.EventBatch)

func (*RowsMetric) Reset

func (rm *RowsMetric) Reset() *RowsMetricState

type RowsMetricState

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

func (*RowsMetricState) Serialize

func (rms *RowsMetricState) Serialize(baseOpts *MeteringOpts) (map[MetricSchema][]string, error)

type SizeBucket

type SizeBucket uint64

type Stats

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

func EmptyStats

func EmptyStats() Stats

func (*Stats) Add

func (s *Stats) Add(value uint64)

func (*Stats) AddBatch

func (s *Stats) AddBatch(count uint64, avgSize uint64)

type Writer

type Writer interface {
	Write(data string) error
	Close() error
}