agent

Documentation

Index

• type Agent
  • func NewAgent(config Config) *Agent
  • func (r *Agent) Run(ctx context.Context, recipe recipe.Recipe) (run Run)
  • func (r *Agent) RunMultiple(ctx context.Context, recipes []recipe.Recipe) []Run
  • func (r *Agent) Validate(rcp recipe.Recipe) []error
• type Config
• type Monitor
• type PluginInfo
• type Run
• type TaskType
• type TimerFn

Types

type Agent

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

Agent runs recipes for specified plugins.

func NewAgent

func NewAgent(config Config) *Agent

NewAgent returns an Agent with plugin factories.

func (*Agent) Run

func (r *Agent) Run(ctx context.Context, recipe recipe.Recipe) (run Run)

Run executes the specified recipe.

func (*Agent) RunMultiple

func (r *Agent) RunMultiple(ctx context.Context, recipes []recipe.Recipe) []Run

RunMultiple executes multiple recipes.

func (*Agent) Validate

func (r *Agent) Validate(rcp recipe.Recipe) []error

Validate checks the recipe for linting errors.

type Config

type Config struct {
	ExtractorFactory     *registry.ExtractorFactory
	ProcessorFactory     *registry.ProcessorFactory
	SinkFactory          *registry.SinkFactory
	Monitor              Monitor
	Logger               log.Logger
	MaxRetries           int
	RetryInitialInterval time.Duration
	StopOnSinkError      bool
	TimerFn              TimerFn
	SinkBatchSize        int
}

type Monitor

type Monitor interface {
	RecordRun(ctx context.Context, run Run)
	RecordPlugin(ctx context.Context, pluginInfo PluginInfo)
	RecordSinkRetryCount(ctx context.Context, pluginInfo PluginInfo)
}

Monitor is the interface for monitoring the agent.

type PluginInfo

type PluginInfo struct {
	RecipeName string
	PluginName string
	PluginType string
	Success    bool
	BatchSize  int
}

type Run

type Run struct {
	Recipe           recipe.Recipe `json:"recipe"`
	Error            error         `json:"error"`
	DurationInMs     int           `json:"duration_in_ms"`
	ExtractorRetries int           `json:"extractor_retries"`
	AssetsExtracted  int           `json:"assets_extracted"`
	RecordCount      int           `json:"record_count"`
	Success          bool          `json:"success"`
}

Run contains the json data

type TaskType

type TaskType string

TaskType is the type of task

const (
	// TaskTypeExtract is the type of task that extracts a record
	TaskTypeExtract TaskType = "extract"
	// TaskTypeProcess is the type of task that processes a record
	TaskTypeProcess TaskType = "process"
	// TaskTypeSink is the type of task that sends a record to a sink
	TaskTypeSink TaskType = "sink"
)

type TimerFn

type TimerFn func() func() int

TimerFn of function type