etcd

Documentation

Overview

Copyright © 2020 Marvin

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright © 2020 Marvin

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright © 2020 Marvin

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright © 2020 Marvin

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Constants
• type Event
• type Executor
  • func NewExecutor(ctx context.Context, etcdClient *clientv3.Client, balanceSleepTime int64) *Executor
  • func (e *Executor) Execute()
  • func (e *Executor) PushTaskPlan(p *Plan)
• type Locker
  • func NewLocker(name string, etcdClient *clientv3.Client) *Locker
  • func (l *Locker) Lock() (err error)
  • func (l *Locker) UnLock() error
• type Plan
  • func (p *Plan) String() string
• type Scheduler
  • func NewScheduler(ctx context.Context, executor *Executor, taskEventChanSize int64) *Scheduler
  • func (s *Scheduler) CreateTaskPlan(eve *Event) (*Plan, error)
  • func (s *Scheduler) HandleTaskEvent(eve *Event)
  • func (s *Scheduler) PushTaskEvent(eve *Event)
  • func (s *Scheduler) ScheduleEvent()
  • func (s *Scheduler) Scheduling()
  • func (s *Scheduler) TryScheduling() (interval time.Duration)
• type Task
  • func (t *Task) String() string
• type Watcher
  • func NewWatcher(ctx context.Context, addr string, scheduler *Scheduler, ...) *Watcher
  • func (w *Watcher) Watch()

Constants

const (
	DefaultScheduledTaskSubmitPrefixKey = "/task/submit/"
	DefaultScheduledTaskKillPrefixKey   = "/task/kill/"
	DefaultScheduledTaskLockPrefixKey   = "/task/lock/"

	DefaultDistributedLockLease = 5

	DefaultTaskActionSubmitEvent = 0
	DefaultTaskActionDeleteEvent = 1
	DefaultTaskActionKillEvent   = 2

	TaskDatabaseStatusWaiting = "WAITING"
	TaskDatabaseStatusRunning = "RUNNING"
	TaskDatabaseStatusStopped = "STOPPED"
	TaskDatabaseStatusFailed  = "FAILED"
	TaskDatabaseStatusSuccess = "SUCCESS"

	TaskModeStructMigrate = "STRUCT_MIGRATE"
)

Task

Types

type Event

type Event struct {
	Opt  int
	Addr string
	Task *Task
}

type Executor

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

func NewExecutor

func NewExecutor(ctx context.Context, etcdClient *clientv3.Client, balanceSleepTime int64) *Executor

func (*Executor) Execute

func (e *Executor) Execute()

func (*Executor) PushTaskPlan

func (e *Executor) PushTaskPlan(p *Plan)

type Locker

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

func NewLocker

func NewLocker(name string, etcdClient *clientv3.Client) *Locker

func (*Locker) Lock

func (l *Locker) Lock() (err error)

func (*Locker) UnLock

func (l *Locker) UnLock() error

type Plan

type Plan struct {
	Addr       string               `json:"addr"`
	Task       *Task                `json:"task"`
	Expr       *cronexpr.Expression `json:"-"`
	Next       time.Time            `json:"next"`     // next schedule time
	Expected   time.Time            `json:"expected"` // expected schedule time
	Real       time.Time            `json:"real"`     // real schedule time
	Status     string               `json:"status"`
	CancelCtx  context.Context      `json:"-"` // task cancel
	CancelFunc context.CancelFunc   `json:"-"`
}

func (*Plan) String

func (p *Plan) String() string

type Scheduler

type Scheduler struct {
	Ctx           context.Context
	Executor      *Executor
	TaskEventChan chan *Event
	TaskPlans     map[string]*Plan
}

func NewScheduler

func NewScheduler(ctx context.Context, executor *Executor, taskEventChanSize int64) *Scheduler

func (*Scheduler) CreateTaskPlan

func (s *Scheduler) CreateTaskPlan(eve *Event) (*Plan, error)

func (*Scheduler) HandleTaskEvent

func (s *Scheduler) HandleTaskEvent(eve *Event)

func (*Scheduler) PushTaskEvent

func (s *Scheduler) PushTaskEvent(eve *Event)

func (*Scheduler) ScheduleEvent

func (s *Scheduler) ScheduleEvent()

func (*Scheduler) Scheduling

func (s *Scheduler) Scheduling()

func (*Scheduler) TryScheduling

func (s *Scheduler) TryScheduling() (interval time.Duration)

TryScheduling used for try schedule example: current time 0s, there are the three tasks，and they are executed once at 1s, 2s and 3s respectively now a b c 0s - 1s - 2s - 3s current only wait 1s

type Task

type Task struct {
	Name    string // task unique logo, when create task, need configure
	Express string // scheduled, if express value is null, the task would immediately execute
}

func (*Task) String

func (t *Task) String() string

type Watcher

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

func NewWatcher

func NewWatcher(ctx context.Context, addr string, scheduler *Scheduler, etcdClient *clientv3.Client) *Watcher

func (*Watcher) Watch

func (w *Watcher) Watch()