workceptor

Oct 16, 2024

Changes in this version

+ type WorkPythonCfg struct

+ Config map[string]interface{}

+ Function string

+ Plugin string

+ WorkType string

+ func (cfg WorkPythonCfg) NewWorker(_ BaseWorkUnitForWorkUnit, w *Workceptor, unitID string, workType string) WorkUnit

+ func (cfg WorkPythonCfg) Run() error

May 7, 2024

Changes in this version

+ var KubeAPIWrapperLock *sync.RWMutex

Mar 21, 2024

Changes in this version

+ func IsCompatibleK8S(kw *KubeUnit, versionStr string) bool

+ func ParseTime(s string) *time.Time

+ func ShouldUseReconnect(kw *KubeUnit) bool

+ type KubeAPIWrapper struct

+ func (ku KubeAPIWrapper) BuildConfigFromFlags(masterURL string, kubeconfigPath string) (*rest.Config, error)

+ func (ku KubeAPIWrapper) Create(ctx context.Context, clientset *kubernetes.Clientset, namespace string, ...) (*corev1.Pod, error)

+ func (ku KubeAPIWrapper) Delete(ctx context.Context, clientset *kubernetes.Clientset, namespace string, ...) error

+ func (ku KubeAPIWrapper) Get(ctx context.Context, clientset *kubernetes.Clientset, namespace string, ...) (*corev1.Pod, error)

+ func (ku KubeAPIWrapper) GetLogs(clientset *kubernetes.Clientset, namespace string, name string, ...) *rest.Request

+ func (ku KubeAPIWrapper) InClusterConfig() (*rest.Config, error)

+ func (ku KubeAPIWrapper) List(ctx context.Context, clientset *kubernetes.Clientset, namespace string, ...) (*corev1.PodList, error)

+ func (ku KubeAPIWrapper) NewClientConfigFromBytes(configBytes []byte) (clientcmd.ClientConfig, error)

+ func (ku KubeAPIWrapper) NewDefaultClientConfigLoadingRules() *clientcmd.ClientConfigLoadingRules

+ func (ku KubeAPIWrapper) NewFakeAlwaysRateLimiter() flowcontrol.RateLimiter

+ func (ku KubeAPIWrapper) NewFakeNeverRateLimiter() flowcontrol.RateLimiter

+ func (ku KubeAPIWrapper) NewForConfig(c *rest.Config) (*kubernetes.Clientset, error)

+ func (ku KubeAPIWrapper) NewNotFound(qualifiedResource schema.GroupResource, name string) *apierrors.StatusError

+ func (ku KubeAPIWrapper) NewSPDYExecutor(config *rest.Config, method string, url *url.URL) (remotecommand.Executor, error)

+ func (ku KubeAPIWrapper) OneTermEqualSelector(k string, v string) fields.Selector

+ func (ku KubeAPIWrapper) StreamWithContext(ctx context.Context, exec remotecommand.Executor, ...) error

+ func (ku KubeAPIWrapper) SubResource(clientset *kubernetes.Clientset, podName string, podNamespace string) *rest.Request

+ func (ku KubeAPIWrapper) UntilWithSync(ctx context.Context, lw cache.ListerWatcher, objType runtime.Object, ...) (*watch.Event, error)

+ func (ku KubeAPIWrapper) Watch(ctx context.Context, clientset *kubernetes.Clientset, namespace string, ...) (watch.Interface, error)

+ type KubeAPIer interface

+ BuildConfigFromFlags func(string, string) (*rest.Config, error)

+ Create func(context.Context, *kubernetes.Clientset, string, *corev1.Pod, ...) (*corev1.Pod, error)

+ Delete func(context.Context, *kubernetes.Clientset, string, string, metav1.DeleteOptions) error

+ Get func(context.Context, *kubernetes.Clientset, string, string, metav1.GetOptions) (*corev1.Pod, error)

+ GetLogs func(*kubernetes.Clientset, string, string, *corev1.PodLogOptions) *rest.Request

+ InClusterConfig func() (*rest.Config, error)

+ List func(context.Context, *kubernetes.Clientset, string, metav1.ListOptions) (*corev1.PodList, error)

+ NewClientConfigFromBytes func([]byte) (clientcmd.ClientConfig, error)

+ NewDefaultClientConfigLoadingRules func() *clientcmd.ClientConfigLoadingRules

+ NewFakeAlwaysRateLimiter func() flowcontrol.RateLimiter

+ NewFakeNeverRateLimiter func() flowcontrol.RateLimiter

+ NewForConfig func(*rest.Config) (*kubernetes.Clientset, error)

+ NewNotFound func(schema.GroupResource, string) *apierrors.StatusError

+ NewSPDYExecutor func(*rest.Config, string, *url.URL) (remotecommand.Executor, error)

+ OneTermEqualSelector func(string, string) fields.Selector

+ StreamWithContext func(context.Context, remotecommand.Executor, remotecommand.StreamOptions) error

+ SubResource func(*kubernetes.Clientset, string, string) *rest.Request

+ UntilWithSync func(context.Context, cache.ListerWatcher, runtime.Object, watch2.PreconditionFunc, ...) (*watch.Event, error)

+ Watch func(context.Context, *kubernetes.Clientset, string, metav1.ListOptions) (watch.Interface, error)

+ var KubeAPIWrapperInstance KubeAPIer

+ type KubeExtraData struct

+ Command string

+ Image string

+ KubeConfig string

+ KubeNamespace string

+ KubePod string

+ Params string

+ PodName string

+ type KubeUnit struct

+ func (kw *KubeUnit) Cancel() error

+ func (kw *KubeUnit) Release(force bool) error

+ func (kw *KubeUnit) Restart() error

+ func (kw *KubeUnit) SetFromParams(params map[string]string) error

+ func (kw *KubeUnit) Start() error

+ func (kw *KubeUnit) Status() *StatusFileData

+ func (kw *KubeUnit) UnredactedStatus() *StatusFileData

type KubeWorkerCfg

+ func (cfg KubeWorkerCfg) NewkubeWorker(bwu BaseWorkUnitForWorkUnit, w *Workceptor, unitID string, workType string, ...) WorkUnit

Feb 7, 2024

Changes in this version

type BaseWorkUnit

+ func (bwu *BaseWorkUnit) GetCancel() context.CancelFunc

+ func (bwu *BaseWorkUnit) GetContext() context.Context

+ func (bwu *BaseWorkUnit) GetStatusCopy() StatusFileData

+ func (bwu *BaseWorkUnit) GetStatusLock() *sync.RWMutex

+ func (bwu *BaseWorkUnit) GetStatusWithoutExtraData() *StatusFileData

+ func (bwu *BaseWorkUnit) GetWorkceptor() *Workceptor

+ func (bwu *BaseWorkUnit) SetStatusExtraData(ed interface{})

+ func (bwu *BaseWorkUnit) SetWorkceptor(w *Workceptor)

+ type BaseWorkUnitForWorkUnit interface

+ CancelContext func()

+ GetCancel func() context.CancelFunc

+ GetContext func() context.Context

+ GetStatusCopy func() StatusFileData

+ GetStatusLock func() *sync.RWMutex

+ GetStatusWithoutExtraData func() *StatusFileData

+ GetWorkceptor func() *Workceptor

+ ID func() string

+ Init func(w *Workceptor, unitID string, workType string, fs FileSystemer, ...)

+ LastUpdateError func() error

+ Load func() error

+ MonitorLocalStatus func()

+ Release func(force bool) error

+ Save func() error

+ SetFromParams func(_ map[string]string) error

+ SetStatusExtraData func(interface{})

+ SetWorkceptor func(*Workceptor)

+ Status func() *StatusFileData

+ StatusFileName func() string

+ StdoutFileName func() string

+ UnitDir func() string

+ UnredactedStatus func() *StatusFileData

+ UpdateBasicStatus func(state int, detail string, stdoutSize int64)

+ UpdateFullStatus func(statusFunc func(*StatusFileData))

+ type CommandExtraData struct

+ Params string

+ Pid int

+ type RemoteExtraData struct

+ Expiration time.Time

+ LocalCancelled bool

+ LocalReleased bool

+ RemoteNode string

+ RemoteParams map[string]string

+ RemoteStarted bool

+ RemoteUnitID string

+ RemoteWorkType string

+ SignWork bool

+ TLSClient string

+ type ServerForWorkceptor interface

+ AddControlFunc func(name string, cType controlsvc.ControlCommandType) error

+ ConnectionListener func(ctx context.Context, listener net.Listener)

+ RunControlSession func(conn net.Conn)

+ RunControlSvc func(ctx context.Context, service string, tlscfg *tls.Config, unixSocket string, ...) error

+ SetServerNet func(n controlsvc.Neter)

+ SetServerTLS func(t controlsvc.Tlser)

+ SetServerUtils func(u controlsvc.Utiler)

+ SetupConnection func(conn net.Conn)

type WorkUnit

+ func NewRemoteWorker(bwu BaseWorkUnitForWorkUnit, w *Workceptor, unitID, workType string) WorkUnit

Oct 13, 2023

Changes in this version

type BaseWorkUnit

+ func (bwu *BaseWorkUnit) CancelContext()

+ func (bwu *BaseWorkUnit) MonitorLocalStatus()

+ type FileReadCloser interface

+ type FileSystem struct

+ func (FileSystem) Open(name string) (*os.File, error)

+ func (FileSystem) OpenFile(name string, flag int, perm os.FileMode) (*os.File, error)

+ func (FileSystem) RemoveAll(path string) error

+ func (FileSystem) Stat(name string) (os.FileInfo, error)

+ type FileSystemer interface

+ Open func(name string) (*os.File, error)

+ OpenFile func(name string, flag int, perm os.FileMode) (*os.File, error)

+ RemoveAll func(path string) error

+ Stat func(name string) (os.FileInfo, error)

+ type FileWriteCloser interface

+ type NetceptorForWorkceptor interface

+ AddWorkCommand func(typeName string, verifySignature bool) error

+ DialContext func(ctx context.Context, node string, service string, tlscfg *tls.Config) (*netceptor.Conn, error)

+ GetClientTLSConfig func(name string, expectedHostName string, ...) (*tls.Config, error)

+ GetLogger func() *logger.ReceptorLogger

+ NodeID func() string

+ type RealWatcher struct

+ func (rw *RealWatcher) Add(name string) error

+ func (rw *RealWatcher) Close() error

+ func (rw *RealWatcher) EventChannel() chan fsnotify.Event

+ type STDinReader struct

+ func NewStdinReader(fs FileSystemer, unitdir string) (*STDinReader, error)

+ func (sr *STDinReader) Done() <-chan struct{}

+ func (sr *STDinReader) Error() error

+ func (sr *STDinReader) Read(p []byte) (n int, err error)

+ func (sr *STDinReader) SetReader(reader FileReadCloser)

+ type STDoutWriter struct

+ func NewStdoutWriter(fs FileSystemer, unitdir string) (*STDoutWriter, error)

+ func (sw *STDoutWriter) SetWriter(writer FileWriteCloser)

+ func (sw *STDoutWriter) Size() int64

+ func (sw *STDoutWriter) Write(p []byte) (n int, err error)

+ type WatcherWrapper interface

+ Add func(name string) error

+ Close func() error

+ EventChannel func() chan fsnotify.Event

May 24, 2023

Changes in this version

+ const WorkStateCanceled

+ var ErrPodFailed = fmt.Errorf("pod failed to start")

+ type CommandWorkerCfg struct

+ AllowRuntimeParams bool

+ Command string

+ Params string

+ VerifySignature bool

+ WorkType string

+ func (cfg CommandWorkerCfg) GetVerifySignature() bool

+ func (cfg CommandWorkerCfg) GetWorkType() string

+ func (cfg CommandWorkerCfg) NewWorker(w *Workceptor, unitID string, workType string) WorkUnit

+ func (cfg CommandWorkerCfg) Run() error

+ type KubeWorkerCfg struct

+ AllowRuntimeAuth bool

+ AllowRuntimeCommand bool

+ AllowRuntimeParams bool

+ AllowRuntimePod bool

+ AuthMethod string

+ Command string

+ DeletePodOnRestart bool

+ Image string

+ KubeConfig string

+ Namespace string

+ Params string

+ Pod string

+ StreamMethod string

+ VerifySignature bool

+ WorkType string

+ func (cfg KubeWorkerCfg) GetVerifySignature() bool

+ func (cfg KubeWorkerCfg) GetWorkType() string

+ func (cfg KubeWorkerCfg) NewWorker(w *Workceptor, unitID string, workType string) WorkUnit

+ func (cfg KubeWorkerCfg) Prepare() error

+ func (cfg KubeWorkerCfg) Run() error

+ type SigningKeyPrivateCfg struct

+ PrivateKey string

+ TokenExpiration string

+ func (cfg SigningKeyPrivateCfg) Prepare() error

+ func (cfg SigningKeyPrivateCfg) PrepareSigningKeyPrivateCfg() (*time.Duration, error)

+ type VerifyingKeyPublicCfg struct

+ PublicKey string

+ func (cfg VerifyingKeyPublicCfg) Prepare() error

+ func (cfg VerifyingKeyPublicCfg) PrepareVerifyingKeyPublicCfg() error

type Workceptor

+ Cancel context.CancelFunc

+ SigningExpiration time.Duration

+ SigningKey string

+ VerifyingKey string

+ type WorkerConfig interface

+ GetVerifySignature func() bool

+ GetWorkType func() string

+ NewWorker func(w *Workceptor, unitID string, workType string) WorkUnit

Dec 3, 2022

Changes in this version

type BaseWorkUnit

+ func (bwu *BaseWorkUnit) Debug(format string, v ...interface{})

+ func (bwu *BaseWorkUnit) Error(format string, v ...interface{})

+ func (bwu *BaseWorkUnit) Info(format string, v ...interface{})

+ func (bwu *BaseWorkUnit) Warning(format string, v ...interface{})

Mar 21, 2022

Changes in this version

+ var ErrImagePullBackOff = fmt.Errorf("container failed to start")

Nov 12, 2021

Changes in this version

+ const MaxWorkSleep

+ const SuccessWorkSleep

+ const WorkStateFailed

+ const WorkStatePending

+ const WorkStateRunning

+ const WorkStateSucceeded

+ var ErrPending = fmt.Errorf("operation pending")

+ var ErrPodCompleted = fmt.Errorf("pod ran to completion")

+ func IsComplete(workState int) bool

+ func IsPending(err error) bool

+ func WorkStateToString(workState int) string

+ type BaseWorkUnit struct

+ func (bwu *BaseWorkUnit) ID() string

+ func (bwu *BaseWorkUnit) Init(w *Workceptor, unitID string, workType string)

+ func (bwu *BaseWorkUnit) LastUpdateError() error

+ func (bwu *BaseWorkUnit) Load() error

+ func (bwu *BaseWorkUnit) Release(force bool) error

+ func (bwu *BaseWorkUnit) Save() error

+ func (bwu *BaseWorkUnit) SetFromParams(params map[string]string) error

+ func (bwu *BaseWorkUnit) Status() *StatusFileData

+ func (bwu *BaseWorkUnit) StatusFileName() string

+ func (bwu *BaseWorkUnit) StdoutFileName() string

+ func (bwu *BaseWorkUnit) UnitDir() string

+ func (bwu *BaseWorkUnit) UnredactedStatus() *StatusFileData

+ func (bwu *BaseWorkUnit) UpdateBasicStatus(state int, detail string, stdoutSize int64)

+ func (bwu *BaseWorkUnit) UpdateFullStatus(statusFunc func(*StatusFileData))

+ type Command struct

+ AllowRuntimeParams bool

+ Command string

+ Params string

+ WorkType string

+ type Kubernetes struct

+ AllowRuntimeAuth bool

+ AllowRuntimeCommand bool

+ AllowRuntimeParams bool

+ AllowRuntimePod bool

+ AuthMethod *string

+ Command string

+ Image string

+ KeepPodOnRestart bool

+ KubeConfig *string

+ Namespace *string

+ Params string

+ Pod string

+ StreamMethod *string

+ WorkType string

+ type NewWorkerFunc func(w *Workceptor, unitID string, workType string) WorkUnit

+ type Python struct

+ Config map[string]interface{}

+ Function string

+ Plugin string

+ WorkType string

+ type StatusFileData struct

+ Detail string

+ ExtraData interface{}

+ State int

+ StdoutSize int64

+ WorkType string

+ func (sfd *StatusFileData) Load(filename string) error

+ func (sfd *StatusFileData) Save(filename string) error

+ func (sfd *StatusFileData) UpdateBasicStatus(filename string, state int, detail string, stdoutSize int64) error

+ func (sfd *StatusFileData) UpdateFullStatus(filename string, statusFunc func(*StatusFileData)) error

+ type WorkUnit interface

+ Cancel func() error

+ ID func() string

+ LastUpdateError func() error

+ Load func() error

+ Release func(force bool) error

+ Restart func() error

+ Save func() error

+ SetFromParams func(params map[string]string) error

+ Start func() error

+ Status func() *StatusFileData

+ StatusFileName func() string

+ StdoutFileName func() string

+ UnitDir func() string

+ UnredactedStatus func() *StatusFileData

+ UpdateBasicStatus func(state int, detail string, stdoutSize int64)

+ UpdateFullStatus func(statusFunc func(*StatusFileData))

+ type Workceptor struct

+ var MainInstance *Workceptor

+ func New(ctx context.Context, nc *netceptor.Netceptor, dataDir string) (*Workceptor, error)

+ func (w *Workceptor) AllocateRemoteUnit(remoteNode, remoteWorkType, tlsClient, ttl string, signWork bool, ...) (WorkUnit, error)

+ func (w *Workceptor) AllocateUnit(workTypeName string, params map[string]string) (WorkUnit, error)

+ func (w *Workceptor) CancelUnit(unitID string) error

+ func (w *Workceptor) GetResults(ctx context.Context, unitID string, startPos int64) (chan []byte, error)

+ func (w *Workceptor) ListKnownUnitIDs() []string

+ func (w *Workceptor) RegisterWithControlService(cs *controlsvc.Server) error

+ func (w *Workceptor) RegisterWorker(typeName string, newWorkerFunc NewWorkerFunc, verifySignature bool) error

+ func (w *Workceptor) ReleaseUnit(unitID string, force bool) error

+ func (w *Workceptor) ShouldVerifySignature(workType string) bool

+ func (w *Workceptor) StartUnit(unitID string) error

+ func (w *Workceptor) UnitStatus(unitID string) (*StatusFileData, error)

+ func (w *Workceptor) VerifySignature(signature string) error

+ type Workers struct

+ Command []Command

+ Kubernetes []Kubernetes

+ Python []Python

+ func (s *Workers) Setup(wc *Workceptor) error