job

Documentation

Overview

Package job provides an interface to a Telemetry Agent job. A job represents the smallest unit of work that the agent recognizes—the specific of what a job does are left to the individual plugin

Index

• func AllPlugins() map[string]PluginFactory
• func GetPluginRegistrationErrors() []error
• func RegisterPlugin(name string, factory PluginFactory)
• type Job
  • func (j *Job) Config() map[string]interface{}
  • func (j *Job) CreateFlow(tag string, variant, sourceProvider, filter, params string) (*gotelemetry.Flow, error)
  • func (j *Job) Debugf(format string, v ...interface{})
  • func (j *Job) GetFlowLayout(id string) (*gotelemetry.Flow, error)
  • func (j *Job) GetFlowTagLayout(tag string) (*gotelemetry.Flow, error)
  • func (j *Job) GetOrCreateBoard(name, prefix string, templateSource string) (*gotelemetry.Board, error)
  • func (j *Job) GetOrCreateFlow(tag, variant string, template interface{}) (*gotelemetry.Flow, error)
  • func (j *Job) Log(v ...interface{})
  • func (j *Job) Logf(format string, v ...interface{})
  • func (j *Job) PostFlowUpdate(flow *gotelemetry.Flow)
  • func (j *Job) PostImmediateFlowUpdate(flow *gotelemetry.Flow) error
  • func (j *Job) QueueDataUpdate(tag string, data interface{}, updateType gotelemetry.BatchType)
  • func (j *Job) ReadFlow(f *gotelemetry.Flow) error
  • func (j *Job) ReportError(err error)
  • func (j *Job) SendNotification(notification gotelemetry.Notification, channelTag string, flowTag string) bool
  • func (j *Job) SetFlowError(tag string, body interface{})
  • func (j *Job) SpawnJob(id string, plugin string, cfg map[string]interface{}) error
• type JobManager
  • func NewJobManager(jobConfig config.ConfigInterface, errorChannel chan error, ...) (*JobManager, error)
• type PluginFactory
  • func GetPlugin(name string) (PluginFactory, error)
• type PluginHelper
  • func NewPluginHelper() *PluginHelper
  • func (e *PluginHelper) AddTaskWithClosure(c PluginHelperClosure, interval time.Duration)
  • func (e *PluginHelper) AddTaskWithClosureForFlowWithTag(c PluginHelperClosureWithFlow, interval time.Duration, ...) error
  • func (e *PluginHelper) AddTaskWithClosureFromBoardForFlowWithTag(c PluginHelperClosureWithFlow, interval time.Duration, b *gotelemetry.Board, ...) error
  • func (e *PluginHelper) AddTaskWithFileObservation(c PluginHelperClosure, path string)
  • func (e *PluginHelper) Reconfigure(job *Job, config map[string]interface{}) error
  • func (e *PluginHelper) Run(job *Job)
  • func (e *PluginHelper) RunOnce(job *Job)
  • func (e *PluginHelper) Terminate(job *Job)
  • func (e *PluginHelper) TrackTime(job *Job, start time.Time, template string)
• type PluginHelperClosure
• type PluginHelperClosureWithFlow
• type PluginInstance

Functions

func AllPlugins

func AllPlugins() map[string]PluginFactory

func GetPluginRegistrationErrors

func GetPluginRegistrationErrors() []error

func RegisterPlugin

func RegisterPlugin(name string, factory PluginFactory)

Types

type Job

type Job struct {
	ID string // The ID of the job
	// contains filtered or unexported fields
}

func (*Job) Config

func (j *Job) Config() map[string]interface{}

Retrieve the configuration data associated with this job

func (*Job) CreateFlow

func (j *Job) CreateFlow(tag string, variant, sourceProvider, filter, params string) (*gotelemetry.Flow, error)

CreateFlow creates a new flow.

func (*Job) Debugf

func (j *Job) Debugf(format string, v ...interface{})

Debugf sends a formatted string to the agent's debug log, if it exists. It works like log.Logf

func (*Job) GetFlowLayout

func (j *Job) GetFlowLayout(id string) (*gotelemetry.Flow, error)

GetFlowLayout returns the layout of a given flow

func (*Job) GetFlowTagLayout

func (j *Job) GetFlowTagLayout(tag string) (*gotelemetry.Flow, error)

func (*Job) GetOrCreateBoard

func (j *Job) GetOrCreateBoard(name, prefix string, templateSource string) (*gotelemetry.Board, error)

GetOrCreateBoard either creates a board based on an exported template, or retrieves it if a board with the same name already exists.

The template must be passed in JSON format as a string (you can use gotelemetry/boarddump to generate a template based on an existing board).

func (*Job) GetOrCreateFlow

func (j *Job) GetOrCreateFlow(tag, variant string, template interface{}) (*gotelemetry.Flow, error)

func (*Job) Log

func (j *Job) Log(v ...interface{})

Log sends data to the agent's global log. It works like log.Log

func (*Job) Logf

func (j *Job) Logf(format string, v ...interface{})

Logf sends a formatted string to the agent's global log. It works like log.Logf

func (*Job) PostFlowUpdate

func (j *Job) PostFlowUpdate(flow *gotelemetry.Flow)

PostFlowUpdate queues a flow update. The method returns immediately, but the update will most likely be sent to the Telemetry API at a later point based on the configuration of the underlying stream

func (*Job) PostImmediateFlowUpdate

func (j *Job) PostImmediateFlowUpdate(flow *gotelemetry.Flow) error

func (*Job) QueueDataUpdate

func (j *Job) QueueDataUpdate(tag string, data interface{}, updateType gotelemetry.BatchType)

PostDataUpdate queues a data update. The update can contain arbitrary data that is sent to the API without any client-side validation.

func (*Job) ReadFlow

func (j *Job) ReadFlow(f *gotelemetry.Flow) error

ReadFlow populates a flow struct with the data that is currently on the server Note that it is not necessary to populate f.Data, as the method will automatically initialize a nil value with the appropriate data structure for the flow's variant.

func (*Job) ReportError

func (j *Job) ReportError(err error)

ReportError sends a formatted error to the agent's global error log. This should be a plugin's preferred error reporting method when running.

func (*Job) SendNotification

func (j *Job) SendNotification(notification gotelemetry.Notification, channelTag string, flowTag string) bool

func (*Job) SetFlowError

func (j *Job) SetFlowError(tag string, body interface{})

func (*Job) SpawnJob

func (j *Job) SpawnJob(id string, plugin string, cfg map[string]interface{}) error

type JobManager

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

func NewJobManager

func NewJobManager(jobConfig config.ConfigInterface, errorChannel chan error, completionChannel chan bool) (*JobManager, error)

type PluginFactory

type PluginFactory func() PluginInstance

func GetPlugin

func GetPlugin(name string) (PluginFactory, error)

type PluginHelper

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

struct PluginHelper simplifies the process of creating plugins by providing most of the required plumbing and allowing the developer to focus on application-specific functionality.

When using PluginHelper as the basis for a plugin, you are only required to provide an Init() method in which you configure one or more tasks, which can optionally be associated with a flow.

PluginHelper will automatically execute tasks asynchronously on a schedule. You can, therefore, consider tasks single-purpose and synchronous, performing whatever functionality you require and then exiting immediately.

func NewPluginHelper

func NewPluginHelper() *PluginHelper

Creates a new plugin helper and returns it

func (*PluginHelper) AddTaskWithClosure

func (e *PluginHelper) AddTaskWithClosure(c PluginHelperClosure, interval time.Duration)

Adds a task to the plugin. The task will be run automarically after the duration specified by the interval parameter. Note that interval is measured starting from the end of the last execution; therefore, you do not need to worry about conditions like slow networking causing successive iterations of a task to “execute over each other.”

func (*PluginHelper) AddTaskWithClosureForFlowWithTag

func (e *PluginHelper) AddTaskWithClosureForFlowWithTag(c PluginHelperClosureWithFlow, interval time.Duration, flows map[string]*gotelemetry.Flow, tag string) error

Adds a task associated with a flow taken from a map of flows. You can obtain a map of flows by calling the MapWidgetsToFlows() method of gotelemetry.Board.

func (*PluginHelper) AddTaskWithClosureFromBoardForFlowWithTag

func (e *PluginHelper) AddTaskWithClosureFromBoardForFlowWithTag(c PluginHelperClosureWithFlow, interval time.Duration, b *gotelemetry.Board, tag string) error

Adds a task associated with a flow taken from a board. This method automatically handles board prefixes; therefore, you must use the tags exactly as they are defined when in the board template.

func (*PluginHelper) AddTaskWithFileObservation

func (e *PluginHelper) AddTaskWithFileObservation(c PluginHelperClosure, path string)

func (*PluginHelper) Reconfigure

func (e *PluginHelper) Reconfigure(job *Job, config map[string]interface{}) error

By default, the plugin helper refuses to reconfigure plugins.

func (*PluginHelper) Run

func (e *PluginHelper) Run(job *Job)

Run method satisfies the requirements of the PluginInstance interface, executing all the tasks asynchronously.

func (*PluginHelper) RunOnce

func (e *PluginHelper) RunOnce(job *Job)

func (*PluginHelper) Terminate

func (e *PluginHelper) Terminate(job *Job)

Terminate waits for all outstanding tasks to be completed and then returns.

func (*PluginHelper) TrackTime

func (e *PluginHelper) TrackTime(job *Job, start time.Time, template string)

TrackTime can be used in a deferred call near the beginning of a function to automatically determine how long that function runs for.

For example:

func test(j *Job) {
	defer plugin.TrackTime(job, time.Now(), "Function test took %s to run.")
}

type PluginHelperClosure

type PluginHelperClosure func(job *Job)

A simple task closure

type PluginHelperClosureWithFlow

type PluginHelperClosureWithFlow func(job *Job, f *gotelemetry.Flow)

A task closure that's associated with a flow

type PluginInstance

type PluginInstance interface {
	Init(job *Job) error                                       // Initializes the instance
	Run(job *Job)                                              // Runs the instance synchronously until Terminate() is called
	RunOnce(job *Job)                                          // Runs the instance synchronously exactly one time and returns immediately
	Reconfigure(job *Job, config map[string]interface{}) error // Dynamically reconfigures the instance, or returns an error if it can't
	Terminate(job *Job)                                        // Terminates the instance, returning only when its execution is complete
}

Interface PluginInstance represents an instance of a plugin that is called upon to perform a job.

In other words, this is what you are supposed to write.