core

package
v0.0.0-...-99a80fa Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 9, 2020 License: MIT Imports: 41 Imported by: 4

Documentation

Overview

Package prometheus provides Prometheus implementations for metrics. Individual metrics are mapped to their Prometheus counterparts, and (depending on the constructor used) may be automatically registered in the global Prometheus metrics registry.

Index

Constants

View Source 
const (
	StateAvaiable = "available"
	// available
	StateExecuting = "executing"
	StateFailed    = "failed"
	StateFinish    = "finish"

	// prometheus
	PrometheusNamespace = "Jelly"
	PrometheusSubsystem = "Schedule"
)
View Source 
const (
	// logger const field
	ProjectKey   = "project"
	ProjectValue = "schedule"
)
View Source 
const DefaultConfigFilename = "/etc/config/schedule_config.yaml"
View Source 
const (
	EmptyJobRequest = "{}"
)
View Source 
const (
	HeadCountKey = "X-Total-Count"
)
View Source 
const (
	JsonRPCPath = `/rpc`
)

把Job封装成一个RPC服务

View Source 
const (
	RemoteServerMethod = `JsonRPCService.Exec`
)

Variables

View Source 
var (
	// etcd
	ErrKeyAlreadyExists  = errors.New("key already exists")
	ErrEtcdLeaseNotFound = errors.New("lease not found")
	ErrJobNotFound       = errors.New("job not found")
	ErrorJobParaInvalid  = errors.New("job para invalid")

	// api
	ErrBadRequest    = newApiError(http.StatusBadRequest, "StatusBadRequest")
	ErrBadCronExpr   = newApiError(http.StatusBadRequest, "ErrBadCronExpr")
	ErrorInvalidPara = newApiError(http.StatusBadRequest, "ErrorInvalidPara")
	ErrNotFound      = newApiError(http.StatusNotFound, "ErrNotFound")
)
View Source 
var (
	JobPrefix = `/schedule/job`
	JobFormat = fasttemplate.New(JobPrefix+`/{Name}`, "{", "}")
	TTL       = int64(10)
)

提供给用户使用, 内部会调用RPC, 抽象成服务 并注册到etcd

View Source 
var (
	EParse       = errors.New("parse pattern failed")
	EPlaceHolder = errors.New("unknown placeholder format")
	EDate        = errors.New("not a date")
	ENumber      = errors.New("not a number")
	EAlpha       = errors.New("not a alpha")
)
View Source 
var (
	WorkflowTableName      = `workflow`
	CreateWorkflowTableDDL = `create table if not exists ` + WorkflowTableName + `
  (
      id                    varchar primary key,
      name                  varchar,
      description           varchar,
			expression            varchar,
			cron                  varchar,
      para                  varchar,
      success_limit         int,
      failed_limit          int,
      belong_executor       varchar,
			state                 varchar,
      create_time           bigint default extract(epoch from now())::bigint,
      update_time           bigint default extract(epoch from now())::bigint
  );`
	WorkflowTableSelectColumn  = `*`
	WorkflowTableColumn        = `id,name,description,expression,cron,para,success_limit,failed_limit,belong_executor,state,create_time,update_time`
	WorkflowTableColumnSize    = len(strings.Split(WorkflowTableColumn, ","))
	WorkflowTableOnConflictDDL = fmt.Sprintf(`
  on conflict (id) 
  do update set
  name = excluded.name,
  description = excluded.description, 
	expression = excluded.expression,
	cron = excluded.cron,
  para = excluded.para,
  success_limit = excluded.success_limit,
  failed_limit = excluded.failed_limit,
  belong_executor = excluded.belong_executor,
  state = excluded.state,
  update_time = GREATEST(%s.update_time, excluded.update_time);`, WorkflowTableName)
)

// 执行几次结束 SuccessLimit int64 `json:"executeLimit" yaml:"executeLimit" ` // 碰到错误的方式 ErrorPolicy string `json:"errorPolicy" yaml:"errorPolicy"` // 可以指定由哪个执行器执行 BelongExecutor string `json:"belongExecutor" yaml:"belongExecutor" `

Functions

func DefaultErrorEncoder 

func DefaultErrorEncoder(err error, w http.ResponseWriter)

func GenJobRequestStringByMeta 

func GenJobRequestStringByMeta(sep string, src ...*JobRequest) (string, error)

func LoadUserConfig 

func LoadUserConfig(field string, obj interface{}) error

func MarshalJobDescription 

func MarshalJobDescription(j *JobDescription) ([]byte, error)

func MarshalJobRequest 

func MarshalJobRequest(req *JobRequest) (string, error)

func MarshalJobRequests 

func MarshalJobRequests(sep string, reqs ...*JobRequest) (string, error)

func NewCronAPI 

func NewCronAPI() *cronAPI

func NewJobAPI 

func NewJobAPI(etcd *Etcd) *jobAPI

func NewPostgres 

func NewPostgres(conf *PostgresConfig) *sqlx.DB

func NewScheduleAPI 

func NewScheduleAPI(etcd *Etcd, db *sqlx.DB, config *HttpConfig) *scheduleAPI

func NewWorkflowAPI 

func NewWorkflowAPI(db *sqlx.DB) *workFlowAPI

func StartClientJob 

func StartClientJob(job Job)

Types

type Counter 

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

Counter implements Counter, via a Prometheus CounterVec.

func NewCounter 

func NewCounter(cv *prometheus.CounterVec) *Counter

NewCounter wraps the CounterVec and returns a usable Counter object.

func NewCounterFrom 

func NewCounterFrom(opts prometheus.CounterOpts, labelNames []string) *Counter

NewCounterFrom constructs and registers a Prometheus CounterVec, and returns a usable Counter object.

func (*Counter) Add 

func (c *Counter) Add(delta float64)

Add implements Counter.

func (*Counter) With 

func (c *Counter) With(labelValues ...string) ICounter

With implements Counter.

type CreateWorkflowRequest 

type CreateWorkflowRequest struct {
	Workflows []*WorkFlow
}

type CreateWorkflowResponse 

type CreateWorkflowResponse struct {
	Ids []string `json:"ids"`
}

type DecodeRequestFunc 

type DecodeRequestFunc func(r *http.Request) (interface{}, error)

type DeleteWorkflowRequest 

type DeleteWorkflowRequest struct {
	Ids []string `json:"ids"`
}

type DeleteWorkflowResponse 

type DeleteWorkflowResponse struct {
}

type Endpoint 

type Endpoint func(ctx context.Context, req interface{}) (resp interface{}, err error)

type Errors 

type Errors []error

func (Errors) Error 

func (e Errors) Error() string

type Etcd 

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

func NewEtcd 

func NewEtcd(config *EtcdConfig) *Etcd

func (*Etcd) Close 

func (e *Etcd) Close() error

func (*Etcd) DelKey 

func (e *Etcd) DelKey(ctx context.Context, key string) error

func (*Etcd) Get 

func (e *Etcd) Get(ctx context.Context, key string) (value []byte, err error)

func (*Etcd) GetWithPrefixKey 

func (e *Etcd) GetWithPrefixKey(ctx context.Context, prefix string) ([]string, []string, error)

func (*Etcd) GrantLease 

func (e *Etcd) GrantLease(ttl int64) (clientv3.LeaseID, error)

func (*Etcd) InsertKV 

func (e *Etcd) InsertKV(ctx context.Context, key, val string, leaseID clientv3.LeaseID) error

func (*Etcd) InsertKVNoExisted 

func (e *Etcd) InsertKVNoExisted(ctx context.Context, key, val string, leaseID clientv3.LeaseID) error

func (*Etcd) KeepaliveWithTTL 

func (e *Etcd) KeepaliveWithTTL(ctx context.Context, key, value string, ttl int64) error

func (*Etcd) RenewLease 

func (e *Etcd) RenewLease(ctx context.Context, id clientv3.LeaseID) error

func (*Etcd) RevokeLease 

func (e *Etcd) RevokeLease(id clientv3.LeaseID) (*clientv3.LeaseRevokeResponse, error)

func (*Etcd) TryLockWithTTL 

func (e *Etcd) TryLockWithTTL(key string, ttl int64) error

func (*Etcd) WatchWithPrefix 

func (e *Etcd) WatchWithPrefix(key string, cb func(*clientv3.Event) error) error

type EtcdConfig 

type EtcdConfig struct {
	Uri     string `yaml:"uri"`
	Timeout int    `json:"timeout" yaml:"timeout" `
}

type Executor 

type Executor struct {
	CheckWorkFlowTicker *time.Ticker
	// contains filtered or unexported fields
}

func NewExecutor 

func NewExecutor(etcd *Etcd, db *sqlx.DB, config ExecutorConfig) *Executor

type ExecutorConfig 

type ExecutorConfig struct {
	Name                  string `json:"name" yaml:"name" `
	CheckWorkFlowInterval int    `json:"checkWorkFlowInterval" yaml:"checkWorkFlowInterval" `
	MetricPort            int    `json:"metricPort" yaml:"metricPort" `
	Separate              string `json:"separate,omitempty" yaml:"separate" `
}

type ExecutorContext 

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

type ExprListener 

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

func (e *Executor) getJob(jobId string) (Job, error) {

func NewExprListener 

func NewExprListener(
	getFn func(string) (Job, error),
	andFn func(left, right Job) Job,
	orFn func(left, right Job) Job,
	loopFn func(left, right Job) Job,
) *ExprListener

func (*ExprListener) EnterANDOR 

func (e *ExprListener) EnterANDOR(c *parser.ANDORContext)

func (*ExprListener) EnterEveryRule 

func (e *ExprListener) EnterEveryRule(ctx antlr.ParserRuleContext)

func (*ExprListener) EnterID 

func (e *ExprListener) EnterID(c *parser.IDContext)

func (*ExprListener) EnterParenthesis 

func (e *ExprListener) EnterParenthesis(c *parser.ParenthesisContext)

func (*ExprListener) EnterStart 

func (e *ExprListener) EnterStart(c *parser.StartContext)

func (*ExprListener) ExitANDOR 

func (e *ExprListener) ExitANDOR(c *parser.ANDORContext)

func (*ExprListener) ExitEveryRule 

func (e *ExprListener) ExitEveryRule(ctx antlr.ParserRuleContext)

func (*ExprListener) ExitID 

func (e *ExprListener) ExitID(c *parser.IDContext)

func (*ExprListener) ExitParenthesis 

func (e *ExprListener) ExitParenthesis(c *parser.ParenthesisContext)

func (*ExprListener) ExitStart 

func (e *ExprListener) ExitStart(c *parser.StartContext)

func (*ExprListener) Pop 

func (e *ExprListener) Pop() Job

func (*ExprListener) VisitErrorNode 

func (e *ExprListener) VisitErrorNode(node antlr.ErrorNode)

func (*ExprListener) VisitTerminal 

func (e *ExprListener) VisitTerminal(node antlr.TerminalNode)

type Gauge 

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

Gauge implements Gauge, via a Prometheus GaugeVec.

func NewGauge 

func NewGauge(gv *prometheus.GaugeVec) *Gauge

NewGauge wraps the GaugeVec and returns a usable Gauge object.

func NewGaugeFrom 

func NewGaugeFrom(opts prometheus.GaugeOpts, labelNames []string) *Gauge

NewGaugeFrom constructs and registers a Prometheus GaugeVec, and returns a usable Gauge object.

func (*Gauge) Add 

func (g *Gauge) Add(delta float64)

Add is supported by Prometheus GaugeVecs.

func (*Gauge) Set 

func (g *Gauge) Set(value float64)

Set implements Gauge.

func (*Gauge) With 

func (g *Gauge) With(labelValues ...string) IGauge

With implements Gauge.

type GetCronRequest 

type GetCronRequest struct {
	Expr      string `json:"expr,omitempty" yaml:"expr" `
	NextCount int    `json:"next_count,omitempty" yaml:"next_count" `
}

type GetCronResponse 

type GetCronResponse struct {
	Expr  string   `json:"expr,omitempty" yaml:"expr" `
	Nexts []string `json:"nexts,omitempty" yaml:"nexts" `
}

type GetJobRequest 

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

type GetJobResponse 

type GetJobResponse struct {
	JobStats []*JobDescription `json:"jobStats"`
}

func NewGetJobResponse 

func NewGetJobResponse() *GetJobResponse

type GetWorkflowRequest 

type GetWorkflowRequest struct {
	Ids []string `json:"ids"`
}

type GetWorkflowResponse 

type GetWorkflowResponse struct {
	Workflows []*WorkFlow `json:"workflows"`
}

type HandleFunc 

type HandleFunc func(w http.ResponseWriter, r *http.Request)

func HandleFuncWrapper 

func HandleFuncWrapper(dec DecodeRequestFunc, e Endpoint, enc encodeResponseFunc) HandleFunc
type Header interface {
	Headers() http.Header
}

type Histogram 

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

Histogram implements Histogram via a Prometheus HistogramVec. The difference between a Histogram and a Summary is that Histograms require predefined quantile buckets, and can be statistically aggregated.

func NewHistogram 

func NewHistogram(hv *prometheus.HistogramVec) *Histogram

NewHistogram wraps the HistogramVec and returns a usable Histogram object.

func NewHistogramFrom 

func NewHistogramFrom(opts prometheus.HistogramOpts, labelNames []string) *Histogram

NewHistogramFrom constructs and registers a Prometheus HistogramVec, and returns a usable Histogram object.

func (*Histogram) Observe 

func (h *Histogram) Observe(value float64)

Observe implements Histogram.

func (*Histogram) With 

func (h *Histogram) With(labelValues ...string) IHistogram

With implements Histogram.

type HttpConfig 

type HttpConfig struct {
	Port      int `json:"port" yaml:"port" `
	PProfPort int `json:"pprofPort" yaml:"pprofPort" `
}

type ICounter 

type ICounter interface {
	With(labelValues ...string) ICounter
	Add(delta float64)
}

// Counter describes a metric that accumulates values monotonically. // An example of a counter is the number of received HTTP requests.

type IGauge 

type IGauge interface {
	With(labelValues ...string) IGauge
	Set(value float64)
	Add(delta float64)
}

Gauge describes a metric that takes specific values over time. An example of a gauge is the current depth of a job queue.

type IHistogram 

type IHistogram interface {
	With(labelValues ...string) IHistogram
	Observe(value float64)
}

Histogram describes a metric that takes repeated observations of the same kind of thing, and produces a statistical summary of those observations, typically expressed as quantiles or buckets. An example of a histogram is HTTP request latencies.

type Job 

type Job interface {
	Name() string
	Exec(ctx context.Context, req string) (resp string, err error)
}

type JobConfig 

type JobConfig struct {
	Ids  map[string]string `json:"ids" yaml:"ids" `
	Host string            `json:"host" yaml:"host" `
}

type JobDescription 

type JobDescription struct {
	Id          string `json:"id"`
	Name        string `json:"name"`
	Host        string `json:"host"`
	Port        int    `json:"port"`
	ServicePath string `json:"servicePath"`
	JobPath     string `json:"jobPath"`
}

func ListJobStats 

func ListJobStats() ([]*JobDescription, error)

func UnMarshalJobDescription 

func UnMarshalJobDescription(buf []byte) (*JobDescription, error)

func (JobDescription) String 

func (w JobDescription) String() string

func (JobDescription) ToJob 

func (w JobDescription) ToJob() Job

type JobError 

type JobError struct {
	Name    string
	Code    int
	Message string
}

Job 执行期间的错误, 用于串行

func NewJobError 

func NewJobError(name string, message string) *JobError

func (*JobError) Error 

func (j *JobError) Error() string

type JobMiddlewareRequest 

type JobMiddlewareRequest struct {
	Name      string
	Parameter []string
}

type JobRequest 

type JobRequest struct {
	Meta map[string]interface{} `json:"meta,omitempty"`
	//Values  []string               `json:"values,omitempty"`
	Values  map[string][]string `json:"values,omitempty"`
	Pattern string              `json:"pattern,omitempty"`
	// contains filtered or unexported fields
}

Meta 每个request负责解释 Values 呈现给job的值域 Pattern 值域表达式, 负责填充值域

//Values []string `json:"values,omitempty"`

func NewJobRequest 

func NewJobRequest() *JobRequest

func NewJobRequestByKey 

func NewJobRequestByKey(key string, src *JobRequest) *JobRequest

func NewJobRequestByMeta 

func NewJobRequestByMeta(src ...*JobRequest) *JobRequest

func UnMarshalJobRequest 

func UnMarshalJobRequest(req string) (*JobRequest, error)

func UnMarshalJobRequests 

func UnMarshalJobRequests(req, sep string) ([]*JobRequest, error)

func (*JobRequest) GetBoolFromMeta 

func (j *JobRequest) GetBoolFromMeta(key string) bool

func (*JobRequest) GetBytesFromMeta 

func (j *JobRequest) GetBytesFromMeta(key string) []byte

func (*JobRequest) GetInt64FromMeta 

func (j *JobRequest) GetInt64FromMeta(key string) int64

func (*JobRequest) GetStringFromMeta 

func (j *JobRequest) GetStringFromMeta(key string) string

type JobResponse 

type JobResponse JobRequest

type JobServer 

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

func NewJobServer 

func NewJobServer(etcd *Etcd, id string, job Job) *JobServer

func (*JobServer) Close 

func (w *JobServer) Close()

func (*JobServer) Stats 

func (w *JobServer) Stats() string

type JsonRPCServer 

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

func (*JsonRPCServer) Close 

func (d *JsonRPCServer) Close()

func (*JsonRPCServer) Start 

func (d *JsonRPCServer) Start() error

type JsonRPCService 

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

func (*JsonRPCService) Exec 

func (j *JsonRPCService) Exec(r *http.Request, arg *string, result *string) error

type KV 

type KV struct {
	Key   string `json:"key,omitempty"`
	Value string `json:"value,omitempty"`
}

type LabelValues 

type LabelValues []string

func (LabelValues) With 

func (lvs LabelValues) With(labelValues ...string) LabelValues

With validates the input, and returns a new aggregate labelValues.

type ListJobRequest 

type ListJobRequest struct{}

type ListJobResponse 

type ListJobResponse struct {
	JobStats []*JobDescription `json:"jobStats"`
}

func NewListJobResponse 

func NewListJobResponse() *ListJobResponse

type ListWorkflowRequest 

type ListWorkflowRequest struct {
	Header      int      `json:"header"`
	Names       []string `json:"names"`
	States      []string `json:"states"`
	StartTime   int64    `json:"startTime"`
	EndTime     int64    `json:"endTime"`
	CurrentPage uint64   `json:"currentPage"`
	PageSize    uint64   `json:"pageSize"`
}

type ListWorkflowResponse 

type ListWorkflowResponse struct {
	Headers   []*KV       `json:"headers,omitempty"`
	Workflows []*WorkFlow `json:"workflows,omitempty"`
}

type LoopJob 

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

func NewLoopJob 

func NewLoopJob(sep string, mergeFn MergeFunc, jobs ...Job) *LoopJob

func (*LoopJob) Append 

func (s *LoopJob) Append(job Job)

func (*LoopJob) Exec 

func (s *LoopJob) Exec(ctx context.Context, req string) (string, error)

func (*LoopJob) Name 

func (s *LoopJob) Name() string

func (*LoopJob) Progress 

func (s *LoopJob) Progress() int

type MergeFunc 

type MergeFunc func(sep string, paras ...string) (string, error)

type Middleware 

type Middleware func(Endpoint) Endpoint

func Instrumenting 

func Instrumenting(latency IHistogram, success, failed ICounter) Middleware

type ParallelJob 

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

func NewParallelJob 

func NewParallelJob(sep string, splitFn SplitFunc, mergeFn MergeFunc, jobs ...Job) *ParallelJob

func (*ParallelJob) Exec 

func (s *ParallelJob) Exec(ctx context.Context, req string) (string, error)

func (*ParallelJob) Name 

func (s *ParallelJob) Name() string

func (*ParallelJob) Progress 

func (s *ParallelJob) Progress() int

type Pattern 

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

func ParsePattern 

func ParsePattern(pattern string) (*Pattern, error)

func (*Pattern) Arrange 

func (p *Pattern) Arrange() []string

func (*Pattern) Map 

func (p *Pattern) Map(keyGen func(...string) string) map[string][]string

func (*Pattern) ToString 

func (p *Pattern) ToString() string

type PlaceHolder 

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

func NewPlaceHolder 

func NewPlaceHolder(str string, begin, end int) (*PlaceHolder, error)

type PostgresConfig 

type PostgresConfig struct {
	Uri     string `yaml:"uri"`
	MaxIdle int    `yaml:"maxIdle" default:"10"`
	MaxOpen int    `yaml:"maxOpen" default:"100"`
}

type ScheduleConfig 

type ScheduleConfig struct {
	Etcd     EtcdConfig
	Postgres PostgresConfig
	Http     HttpConfig
	Executor ExecutorConfig
	Job      JobConfig
}

func LoadScheduleConfig 

func LoadScheduleConfig(filename string) (*ScheduleConfig, error)

type SerialJob 

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

func NewSerialJob 

func NewSerialJob(sep string, jobs ...Job) *SerialJob

func (*SerialJob) Append 

func (s *SerialJob) Append(job Job)

func (*SerialJob) Exec 

func (s *SerialJob) Exec(ctx context.Context, req string) (string, error)

func (*SerialJob) Name 

func (s *SerialJob) Name() string

func (*SerialJob) Progress 

func (s *SerialJob) Progress() int

type SplitFunc 

type SplitFunc func(sep string, paras string) ([]string, error)

type Stack 

type Stack []string

type StatusCoder 

type StatusCoder interface {
	StatusCode() int
}

type Summary 

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

Summary implements Histogram, via a Prometheus SummaryVec. The difference between a Summary and a Histogram is that Summaries don't require predefined quantile buckets, but cannot be statistically aggregated.

func NewSummary 

func NewSummary(sv *prometheus.SummaryVec) *Summary

NewSummary wraps the SummaryVec and returns a usable Summary object.

func NewSummaryFrom 

func NewSummaryFrom(opts prometheus.SummaryOpts, labelNames []string) *Summary

NewSummaryFrom constructs and registers a Prometheus SummaryVec, and returns a usable Summary object.

func (*Summary) Observe 

func (s *Summary) Observe(value float64)

Observe implements Histogram.

func (*Summary) With 

func (s *Summary) With(labelValues ...string) IHistogram

With implements Histogram.

type SyncMap 

type SyncMap struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewSyncMap 

func NewSyncMap() *SyncMap

func (*SyncMap) Get 

func (s *SyncMap) Get(k string) (interface{}, bool)

func (*SyncMap) Put 

func (s *SyncMap) Put(k string, v interface{})

type SyncRequest 

type SyncRequest struct {
	SrcHost   string
	SrcFiles  []string
	DestHost  string
	DestFiles []string
}

type UpdateWorkflowRequest 

type UpdateWorkflowRequest struct {
	WorkFlows []*WorkFlow `json:"workflows"`
}

type UpdateWorkflowResponse 

type UpdateWorkflowResponse struct {
}

type WorkFlow 

type WorkFlow struct {
	Id          string `json:"id,omitempty"`
	Name        string `json:"name,omitempty"`
	Description string `json:"description,omitempty"`
	Expression  string `json:"expression,omitempty"`
	Cron        string `json:"cron,omitempty"`
	Para        string `json:"para"`

	ExecutorWhenDeployed      bool  `json:"executor_when_deployed" yaml:"executor_when_deployed" `
	ExecutorWhenDeployedDelay int64 `json:"executor_when_deployed_delay" yaml:"executor_when_deployed_delay" `

	// 执行几次结束
	SuccessLimit int64 `json:"successLimit" yaml:"successLimit" `
	// 碰到错误的方式
	FailedLimit int64 `json:"failedLimit" yaml:"failedLimit"`
	// 可以指定由哪个执行器执行
	BelongExecutor string `json:"belongExecutor" yaml:"belongExecutor" `
	State          string `json:"state,omitempty"`
	CreateTime     int64  `json:"createTime,omitempty"`
	UpdateTime     int64  `json:"updateTime,omitempty"`
}

type WorkFlowStatus 

type WorkFlowStatus struct {
	Id                  string
	Executing           bool
	MaxExecuteCount     int64
	SuccessExecuteCount int64
	FailedExecuteCount  int64
	LastExecuteDuration int64
}

// 主要记录workflow runtime 状态记录的

type WorkerRole 

type WorkerRole uint8
const (
	Wait WorkerRole = iota
	Follower
	Leader
)

type WrapperJob 

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

executor从workflow中得到了job的id 利用这个类, 封装成一个Job接口

func NewWrapperJob 

func NewWrapperJob(info *JobDescription) *WrapperJob

func (*WrapperJob) Exec 

func (e *WrapperJob) Exec(ctx context.Context, req string) (string, error)

func (*WrapperJob) Name 

func (e *WrapperJob) Name() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.