core

Documentation

Index

• Variables
• func Cleanup(record utils.Record)
• func GetWorkQueueElements(rname string) []utils.Record
• func InWorkQueue(rname string) bool
• func InitCouch(couchUrl, dbName string)
• func Process(record utils.Record)
• type BlockPolicy
  • func (b *BlockPolicy) Split() []couchdb.CouchDoc
  • func (b *BlockPolicy) Validate() bool
• type Dispatcher
  • func NewDispatcher(maxWorkers, maxQueue int, mfile string, minterval int64) *Dispatcher
  • func (d *Dispatcher) Run(rtype string, interval, cleanup int64)
• type Job
• type Mask
• type Metrics
  • func (m *Metrics) String() string
  • func (m *Metrics) ToDict() map[string]int64
• type MonteCarloPolicy
  • func (b *MonteCarloPolicy) Split() []couchdb.CouchDoc
  • func (b *MonteCarloPolicy) Validate() bool
• type Policy
• type ResubmitBlockPolicy
  • func (b *ResubmitBlockPolicy) Split() []couchdb.CouchDoc
  • func (b *ResubmitBlockPolicy) Validate() bool
• type WorkQueueElement
  • func (w WorkQueueElement) String() string
• type Worker
  • func NewWorker(wid int, jobPool chan chan Job) Worker
  • func (w Worker) Start()
  • func (w Worker) Stop()
• type WorkqueueStatus

Variables

var Client *couchdb.Client

Client points to CouchDB client

var DB couchdb.DatabaseService

DB points to CouchDB

var JobQueue chan Job

JobQueue is a buffered channel that we can send work requests on.

Functions

func Cleanup

func Cleanup(record utils.Record)

Cleanup performs clean-up of WorkQueue

func GetWorkQueueElements

func GetWorkQueueElements(rname string) []utils.Record

GetWorkQueueElements returns list of request in WorkQueue

func InWorkQueue

func InWorkQueue(rname string) bool

InWorkQueue returns if given request name in WorkQueue

func InitCouch

func InitCouch(couchUrl, dbName string)

InitCouch initializes connection to CouchDB

func Process

func Process(record utils.Record)

Process given request

Types

type BlockPolicy

type BlockPolicy struct {
	Name   string
	Record utils.Record
	Config utils.Record
}

BlockPolicy defines block policy

func (*BlockPolicy) Split

func (b *BlockPolicy) Split() []couchdb.CouchDoc

Split method satisfy Policy interface

func (*BlockPolicy) Validate

func (b *BlockPolicy) Validate() bool

Split method satisfy Policy interface

type Dispatcher

type Dispatcher struct {
	// A pool of workers channels that are registered with the dispatcher
	JobPool    chan chan Job
	MaxWorkers int
}

Dispatcher implementation

func NewDispatcher

func NewDispatcher(maxWorkers, maxQueue int, mfile string, minterval int64) *Dispatcher

NewDispatcher returns new instance of Dispatcher type

func (*Dispatcher) Run

func (d *Dispatcher) Run(rtype string, interval, cleanup int64)

Run function starts the worker and dispatch it as go-routine

type Job

type Job struct {
	Request utils.Record
	Type    string
}

Job represents the job to be run with given request

type Mask

type Mask struct {
	InclusiveMask bool
	FirstEvent    int64
	LastEvent     int64
	FirstLumi     int
	LastLumi      int
	FirstRun      int
	LstRun        int
	RunAndLumis   map[int][]int
}

Mask data structure keeps track of run-lumi

type Metrics

type Metrics struct {
	Jobs metrics.Counter // number of jobs
}

Metrics of the agent

var WorkqueueMetrics Metrics

WorkqueueMetrics defines various metrics about the agent work

func (*Metrics) String

func (m *Metrics) String() string

String representation of Metrics

func (*Metrics) ToDict

func (m *Metrics) ToDict() map[string]int64

ToDict converts Metrics structure to a map

type MonteCarloPolicy

type MonteCarloPolicy struct {
	Name   string
	Record utils.Record
	Config utils.Record
}

MonteCarloPolicy defines block policy

func (*MonteCarloPolicy) Split

func (b *MonteCarloPolicy) Split() []couchdb.CouchDoc

Split method satisfy Policy interface

func (*MonteCarloPolicy) Validate

func (b *MonteCarloPolicy) Validate() bool

Split method satisfy Policy interface

type Policy

type Policy interface {
	Split() []WorkQueueElement
	Validate() bool
}

Policy interface defines policy methods

type ResubmitBlockPolicy

type ResubmitBlockPolicy struct {
	Name   string
	Record utils.Record
	Config utils.Record
}

ResubmitBlockPolicy defines block policy

func (*ResubmitBlockPolicy) Split

func (b *ResubmitBlockPolicy) Split() []couchdb.CouchDoc

Split method satisfy Policy interface

func (*ResubmitBlockPolicy) Validate

func (b *ResubmitBlockPolicy) Validate() bool

Split method satisfy Policy interface

type WorkQueueElement

type WorkQueueElement struct {
	couchdb.Document
	Inputs          map[string][]string // {blockName:sites}
	ParentData      map[string][]string // {blockName:sites}
	PileupData      map[string][]string // {blockName:sites}
	ParentFlag      bool
	NumberOfLumis   int
	NumberOfFiles   int
	NumberOfEvents  int
	Jobs            int
	OpenForNewData  bool
	NoInputUpdate   bool
	NoPileupUpdate  bool
	WMSpec          string
	Mask            map[string]int
	BlowupFactor    int
	ACDC            string
	RequestName     string
	TaskName        string
	Dbs             string
	SiteWhiteList   []string
	SiteBlackList   []string
	Priority        int
	ParentQueueUrl  string
	ChildQueueUrl   string
	PercentSuccess  float32
	PercentComplete float32
	WMBSUrl         string
	FilesProcessed  int
	StartPolicy     string
	EndPolicy       string
}

WorkQueueElement structure

func (WorkQueueElement) String

func (w WorkQueueElement) String() string

String function implements Stringer interface

type Worker

type Worker struct {
	Id         int
	JobPool    chan chan Job
	JobChannel chan Job
	// contains filtered or unexported fields
}

Worker represents the worker that executes the job

func NewWorker

func NewWorker(wid int, jobPool chan chan Job) Worker

NewWorker return a new instance of the Worker type

func (Worker) Start

func (w Worker) Start()

Start method starts the run loop for the worker, listening for a quit channel in case we need to stop it

func (Worker) Stop

func (w Worker) Stop()

Stop signals the worker to stop listening for work requests.

type WorkqueueStatus

type WorkqueueStatus struct {
	Addrs            []string         `json:"addrs"`     // list of all IP addresses
	TimeStamp        int64            `json:"ts"`        // time stamp
	Metrics          map[string]int64 `json:"metrics"`   // workqueue metrics
	NumberOfRequests int              `json:"nRequests"` // number of requests in workqueue
}

WorkqueueStatus data type