actionagent

Documentation

Index

• Constants
• type Agent
  • func (agent *Agent) AppendError(err error)
  • func (agent *Agent) Callback()
  • func (agent *Agent) CheckForBreakpointOnFailure(breakPointFile string)
  • func (agent *Agent) Execute(r io.Reader)
  • func (agent *Agent) Exit()
  • func (agent *Agent) ListenSignal()
  • func (agent *Agent) MergeErrors() taskerror.OrderedErrors
  • func (agent *Agent) PreStop()
  • func (agent *Agent) SetCallbackReporter()
  • func (agent *Agent) SetTokenForBootstrap() error
  • func (agent *Agent) Teardown(exitCode ...int)
• type AgentArg
  • func NewAgentArgForPull(pipelineID, pipelineTaskID uint64) *AgentArg
• type Callback
  • func (c *Callback) AppendMetadataFields(fields []*metadata.MetadataField)
  • func (c *Callback) HandleMetaFile(b []byte) error
• type CallbackReporter
• type CenterCallbackReporter
  • func (cr *CenterCallbackReporter) CallbackToPipelinePlatform(cbReq apistructs.PipelineCallbackRequest) error
  • func (cr *CenterCallbackReporter) GetBootstrapInfo(pipelineID, taskID uint64) (apistructs.PipelineTaskGetBootstrapInfoResponse, error)
  • func (cr *CenterCallbackReporter) GetCmsFile(uuid string, absPath string) error
  • func (cr *CenterCallbackReporter) PushCollectorLog(logLines *[]apistructs.LogPushLine) error
  • func (cr *CenterCallbackReporter) SetCollectorAddress(address string)
  • func (cr *CenterCallbackReporter) SetOpenApiToken(token string)
  • func (cr *CenterCallbackReporter) UploadFile(pipelineID, taskID uint64, file *os.File) (*pb.FileUploadResponse, error)
• type EasyUse
• type EdgeCallbackReporter
  • func (er *EdgeCallbackReporter) CallbackToPipelinePlatform(cbReq apistructs.PipelineCallbackRequest) error
  • func (er *EdgeCallbackReporter) GetBootstrapInfo(pipelineID, taskID uint64) (apistructs.PipelineTaskGetBootstrapInfoResponse, error)
  • func (er *EdgeCallbackReporter) GetCmsFile(uuid string, absPath string) error
  • func (er *EdgeCallbackReporter) PushCollectorLog(logLines *[]apistructs.LogPushLine) error
  • func (er *EdgeCallbackReporter) SetCollectorAddress(address string)
  • func (er *EdgeCallbackReporter) SetOpenApiToken(token string)
  • func (er *EdgeCallbackReporter) UploadFile(pipelineID, taskID uint64, file *os.File) (*pb.FileUploadResponse, error)
• type RunningEnvironment

Constants

const (
	EnvContextDir = "CONTEXTDIR"
	EnvWorkDir    = "WORKDIR"
	EnvMetaFile   = "METAFILE"
	EnvUploadDir  = "UPLOADDIR"

	EnvPrefix            = "ACTIONAGENT_"
	EnvActionParamPrefix = "ACTION_"
)

const (
	EnvDiceOpenapiPublicUrl = "DICE_OPENAPI_PUBLIC_URL"
	EnvDiceOpenapiAddr      = "DICE_OPENAPI_ADDR"
	EnvDiceIsEdge           = "DICE_IS_EDGE"
)

const (
	EnvStdErrRegexpList = "ACTIONAGENT_STDERR_REGEXP_LIST"
	EnvMaxCacheFileMB   = "ACTIONAGENT_MAX_CACHE_FILE_MB"
	EnvDefaultShell     = "ACTIONAGENT_DEFAULT_SHELL"
	EnvDefaultTimezone  = "ACTIONAGENT_DEFAULT_TIMEZONE"
)

const (
	EnvSuffixPublicURL = "_PUBLIC_URL"
	EnvSuffixAddr      = "_ADDR"
)

const (
	EnvEnablePushLog2Collector = "ACTIONAGENT_ENABLE_PUSH_LOG_TO_COLLECTOR"
	EnvCollectorAddr           = "COLLECTOR_ADDR"
	EnvTaskLogID               = "TERMINUS_DEFINE_TAG"

	TagDiceOrgName = "dice_org_name"
	TagDiceOrgID   = "dice_org_id"
)

const EncryptedValueMinLen = "ACTIONAGENT_ENCRYPTED_VAlUE_MIN_LEN"

const (
	EnvFileStreamTimeoutSec = "ACTIONAGENT_FILE_STREAM_TIMEOUT_SEC"
)

const (
	MetadataTypeDiceFile = "DiceFile"
)

Types

type Agent

type Agent struct {
	Arg *AgentArg

	EasyUse EasyUse

	// errs collect occurred errors
	Errs []error

	FileWatcher *filewatch.Watcher
	FWLock      sync.Mutex

	// PushedMetaFileMap store key value already pushed
	// first-level key: meta name; second-level key: meta value
	// for situation that one key has multiple values
	PushedMetaFileMap     map[string]map[string]struct{}
	LockPushedMetaFileMap sync.RWMutex

	Ctx      context.Context
	Cancel   context.CancelFunc // cancel when logic done
	ExitCode int

	StdErrRegexpList   []*regexp.Regexp
	MaxCacheFileSizeMB datasize.ByteSize

	CallbackReporter

	TextBlackList []string // enciphered data will Replaced by '******' when log output
}

func (*Agent) AppendError

func (agent *Agent) AppendError(err error)

func (*Agent) Callback

func (agent *Agent) Callback()

Callback .

func (*Agent) CheckForBreakpointOnFailure

func (agent *Agent) CheckForBreakpointOnFailure(breakPointFile string)

CheckForBreakpointOnFailure if step up breakpoint on failure waiting breakpointExitPostFile to be written

func (*Agent) Execute

func (agent *Agent) Execute(r io.Reader)

func (*Agent) Exit

func (agent *Agent) Exit()

Exit exit agent.

func (*Agent) ListenSignal

func (agent *Agent) ListenSignal()

func (*Agent) MergeErrors

func (agent *Agent) MergeErrors() taskerror.OrderedErrors

func (*Agent) PreStop

func (agent *Agent) PreStop()

func (*Agent) SetCallbackReporter

func (agent *Agent) SetCallbackReporter()

func (*Agent) SetTokenForBootstrap

func (agent *Agent) SetTokenForBootstrap() error

func (*Agent) Teardown

func (agent *Agent) Teardown(exitCode ...int)

Teardown teardown agent, including prestop, callback.

type AgentArg

type AgentArg struct {
	PullBootstrapInfo bool `json:"pullBootstrapInfo"`

	Shell     string                   `json:"shell,omitempty"`
	ShellArgs []string                 `json:"shellArgs,omitempty"`
	Commands  interface{}              `json:"commands,omitempty"` // custom action commands -> script -> run
	Context   spec.PipelineTaskContext `json:"context,omitempty"`  // 上下文

	PrivateEnvs map[string]string `json:"privateEnvs,omitempty"`

	PipelineID     uint64 `json:"pipelineID"`
	PipelineTaskID uint64 `json:"pipelineTaskID"`

	EncryptSecretKeys []string `json:"encryptSecretKeys"` // Encrypt envs, it will Replaced by '******' when log output

	// breakpoint config
	DebugOnFailure bool           `json:"debugOnFailure"`
	DebugTimeout   *time.Duration `json:"debugTimeout"`
}

func NewAgentArgForPull

func NewAgentArgForPull(pipelineID, pipelineTaskID uint64) *AgentArg

type Callback

type Callback apistructs.ActionCallback

func (*Callback) AppendMetadataFields

func (c *Callback) AppendMetadataFields(fields []*metadata.MetadataField)

append fields to metadata and limit metadataField

func (*Callback) HandleMetaFile

func (c *Callback) HandleMetaFile(b []byte) error

1) decode as: metadata.Metadata 2) decode as: line(k=v)

type CallbackReporter

type CallbackReporter interface {
	CallbackToPipelinePlatform(cbReq apistructs.PipelineCallbackRequest) error
	UploadFile(pipelineID, taskID uint64, file *os.File) (*pb.FileUploadResponse, error)
	GetBootstrapInfo(pipelineID, taskID uint64) (apistructs.PipelineTaskGetBootstrapInfoResponse, error)
	GetCmsFile(uuid string, absPath string) error
	SetOpenApiToken(token string)
	SetCollectorAddress(address string)
	PushCollectorLog(logLines *[]apistructs.LogPushLine) error
}

type CenterCallbackReporter

type CenterCallbackReporter struct {
	OpenAPIAddr          string
	OpenAPIToken         string
	TokenForBootstrap    string
	CollectorAddr        string
	FileStreamTimeoutSec time.Duration
}

func (*CenterCallbackReporter) CallbackToPipelinePlatform

func (cr *CenterCallbackReporter) CallbackToPipelinePlatform(cbReq apistructs.PipelineCallbackRequest) error

func (*CenterCallbackReporter) GetBootstrapInfo

func (cr *CenterCallbackReporter) GetBootstrapInfo(pipelineID, taskID uint64) (apistructs.PipelineTaskGetBootstrapInfoResponse, error)

func (*CenterCallbackReporter) GetCmsFile

func (cr *CenterCallbackReporter) GetCmsFile(uuid string, absPath string) error

func (*CenterCallbackReporter) PushCollectorLog

func (cr *CenterCallbackReporter) PushCollectorLog(logLines *[]apistructs.LogPushLine) error

func (*CenterCallbackReporter) SetCollectorAddress

func (cr *CenterCallbackReporter) SetCollectorAddress(address string)

func (*CenterCallbackReporter) SetOpenApiToken

func (cr *CenterCallbackReporter) SetOpenApiToken(token string)

func (*CenterCallbackReporter) UploadFile

func (cr *CenterCallbackReporter) UploadFile(pipelineID, taskID uint64, file *os.File) (*pb.FileUploadResponse, error)

type EasyUse

type EasyUse struct {
	ContainerContext  string // 容器内程序运行时上下文目录
	ContainerWd       string // 执行 run 程序时所在目录
	ContainerMetaFile string // metadata 文件

	ContainerUploadDir        string // uploadDir，该目录下的文件在执行结束后会自动上传，并提供下载
	ContainerTempTarUploadDir string // temp tar dir，需要 prepare 时预先创建，用于存放 upload 生成的 tar

	IsEdgeCluster  bool // is edge cluster
	IsEdgePipeline bool // is running on edge pipeline

	RunScript              string // run 文件
	CommandScript          string // custom command script
	RunProcess             *os.Process
	RunMultiStdoutFilePath string   // multiWriter(os.Stdout) 的文件路径
	RunMultiStdout         *os.File // multiWriter(os.Stdout) 的文件
	RunMultiStderrFilePath string   // multiWriter(os.Stderr) 的文件路径
	RunMultiStderr         *os.File // multiWriter(os.Stderr) 的文件
	FlagEndLineForTail     string   // flag end line for tail to stop

	OpenAPIAddr       string
	OpenAPIToken      string
	TokenForBootstrap string
	PipelineAddr      string

	EnablePushLog2Collector bool   // 是否推送日志到 collector
	CollectorAddr           string // collector 地址
	TaskLogID               string // 日志 ID，推送和查询时需要一致

	// Machine stat
	MachineStat taskinspect.PipelineTaskMachineStat

	FileStreamTimeoutSec time.Duration // download or upload file stream timeout
}

type EdgeCallbackReporter

type EdgeCallbackReporter struct {
	PipelineAddr         string
	OpenAPIToken         string
	TokenForBootstrap    string
	CollectorAddr        string
	FileStreamTimeoutSec time.Duration
}

func (*EdgeCallbackReporter) CallbackToPipelinePlatform

func (er *EdgeCallbackReporter) CallbackToPipelinePlatform(cbReq apistructs.PipelineCallbackRequest) error

func (*EdgeCallbackReporter) GetBootstrapInfo

func (er *EdgeCallbackReporter) GetBootstrapInfo(pipelineID, taskID uint64) (apistructs.PipelineTaskGetBootstrapInfoResponse, error)

func (*EdgeCallbackReporter) GetCmsFile

func (er *EdgeCallbackReporter) GetCmsFile(uuid string, absPath string) error

func (*EdgeCallbackReporter) PushCollectorLog

func (er *EdgeCallbackReporter) PushCollectorLog(logLines *[]apistructs.LogPushLine) error

func (*EdgeCallbackReporter) SetCollectorAddress

func (er *EdgeCallbackReporter) SetCollectorAddress(address string)

func (*EdgeCallbackReporter) SetOpenApiToken

func (er *EdgeCallbackReporter) SetOpenApiToken(token string)

func (*EdgeCallbackReporter) UploadFile

func (er *EdgeCallbackReporter) UploadFile(pipelineID, taskID uint64, file *os.File) (*pb.FileUploadResponse, error)

type RunningEnvironment

type RunningEnvironment struct {
	HostIP string
}