sink

Documentation

Index

• type JobRunRow
• type JobWriter
  • func NewJobWriter(path string) (*JobWriter, error)
  • func (j *JobWriter) Close(ctx *armadacontext.Context)
  • func (j *JobWriter) Update(st *model.StateTransition) error
• type NullSink
  • func (s NullSink) Close(ctx *armadacontext.Context)
  • func (s NullSink) OnCycleEnd(_ time.Time, _ *scheduling.SchedulerResult) error
  • func (s NullSink) OnNewStateTransitions(_ []*model.StateTransition) error
• type ParquetSink
  • func NewParquetSink(outputDir string) (*ParquetSink, error)
  • func (s *ParquetSink) Close(ctx *armadacontext.Context)
  • func (s *ParquetSink) OnCycleEnd(time time.Time, result *scheduling.SchedulerResult) error
  • func (s *ParquetSink) OnNewStateTransitions(transitions []*model.StateTransition) error
• type QueueStatsRow
• type QueueStatsWriter
  • func NewQueueStatsWriter(path string) (*QueueStatsWriter, error)
  • func (j *QueueStatsWriter) Close(ctx *armadacontext.Context)
  • func (j *QueueStatsWriter) Update(time time.Time, result *scheduling.SchedulerResult) error
• type Sink

Types

type JobRunRow

type JobRunRow struct {
	Queue            string  `parquet:"name=queue, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	JobSet           string  `parquet:"name=job_set, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	JobId            string  `parquet:"name=job_id, type=BYTE_ARRAY, convertedtype=UTF8"`
	RunId            string  `parquet:"name=run_id, type=BYTE_ARRAY, convertedtype=UTF8"`
	PriorityClass    string  `parquet:"name=priority_class, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	Cpu              float64 `parquet:"name=cpu, type=DOUBLE"`
	Memory           float64 `parquet:"name=memory, type=DOUBLE"`
	Gpu              float64 `parquet:"name=gpu, type=DOUBLE"`
	EphemeralStorage float64 `parquet:"name=ephemeral_storage, type=DOUBLE"`
	ExitCode         int     `parquet:"name=exit_code, type=INT32"`
	State            string  `parquet:"name=state, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	SubmittedTime    int64   `parquet:"name=submitted_time, type=INT64"`
	ScheduledTime    int64   `parquet:"name=scheduled_time, type=INT64"`
	FinishedTime     int64   `parquet:"name=finished_time, type=INT64"`
}

type JobWriter

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

func NewJobWriter

func NewJobWriter(path string) (*JobWriter, error)

func (*JobWriter) Close

func (j *JobWriter) Close(ctx *armadacontext.Context)

func (*JobWriter) Update

func (j *JobWriter) Update(st *model.StateTransition) error

type NullSink

type NullSink struct{}

func (NullSink) Close

func (s NullSink) Close(ctx *armadacontext.Context)

func (NullSink) OnCycleEnd

func (s NullSink) OnCycleEnd(_ time.Time, _ *scheduling.SchedulerResult) error

func (NullSink) OnNewStateTransitions

func (s NullSink) OnNewStateTransitions(_ []*model.StateTransition) error

type ParquetSink

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

func NewParquetSink

func NewParquetSink(outputDir string) (*ParquetSink, error)

func (*ParquetSink) Close

func (s *ParquetSink) Close(ctx *armadacontext.Context)

func (*ParquetSink) OnCycleEnd

func (s *ParquetSink) OnCycleEnd(time time.Time, result *scheduling.SchedulerResult) error

func (*ParquetSink) OnNewStateTransitions

func (s *ParquetSink) OnNewStateTransitions(transitions []*model.StateTransition) error

type QueueStatsRow

type QueueStatsRow struct {
	Ts                int64   `parquet:"name=ts, type=INT64"`
	Queue             string  `parquet:"name=queue, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	Pool              string  `parquet:"name=pool, type=BYTE_ARRAY, convertedtype=UTF8"`
	FairShare         float64 `parquet:"name=fair_share, type=DOUBLE"`
	AdjustedFairShare float64 `parquet:"name=adjusted_fair_share, type=DOUBLE"`
	ActualShare       float64 `parquet:"name=actual_share, type=DOUBLE"`
	CpuShare          float64 `parquet:"name=cpu_share, type=DOUBLE"`
	MemoryShare       float64 `parquet:"name=memory_share, type=DOUBLE"`
	GpuShare          float64 `parquet:"name=gpu_share, type=DOUBLE"`
	AllocatedCPU      int     `parquet:"name=allocated_cpu, type=INT64"`
	AllocatedMemory   int     `parquet:"name=allocated_memory, type=INT64"`
	AllocatedGPU      int     `parquet:"name=allocated_gpu, type=INT64"`
	NumScheduled      int     `parquet:"name=num_scheduled, type=INT32"`
	NumPreempted      int     `parquet:"name=num_preempted, type=INT32"`
	NumEvicted        int     `parquet:"name=num_evicted, type=INT32"`
}

type QueueStatsWriter

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

func NewQueueStatsWriter

func NewQueueStatsWriter(path string) (*QueueStatsWriter, error)

func (*QueueStatsWriter) Close

func (j *QueueStatsWriter) Close(ctx *armadacontext.Context)

func (*QueueStatsWriter) Update

func (j *QueueStatsWriter) Update(time time.Time, result *scheduling.SchedulerResult) error

type Sink

type Sink interface {
	OnNewStateTransitions(transitions []*model.StateTransition) error
	OnCycleEnd(time time.Time, result *scheduling.SchedulerResult) error
	Close(ctx *armadacontext.Context)
}