module

Documentation

Index

• Constants
• Variables
• func DecCountOfRunningGoRoutine()
• func IncCountOfRunningGoRoutine()
• func IncCountOfTask()
• func IncCountOfTaskError()
• func IncCountOfTaskFinished()
• func NewStatus(id string, status string, info string, updateTime time.Time) *config.Status
• type DmPrometheus
  • func NewDmPrometheus(pc *prometheus.Config, onOff chan bool) *DmPrometheus
  • func (dp *DmPrometheus) Run()
  • func (dp *DmPrometheus) Stop()
• type Manager
  • func NewManager(controllers map[string]*chan bool, tasks *TaskQueue, config *config.Config, ...) *Manager
  • func (m *Manager) AddController(taskID string) *chan bool
  • func (m *Manager) Consume()
  • func (m *Manager) HandleJob(job *config.Job, mutex *concurrency.Mutex) error
  • func (m *Manager) StopJob(service *Service) error
• type Service
  • func NewService(job *config.Job, config *config.Config, controller *chan bool, idType string, ...) *Service
  • func (s *Service) Run() (string, error)
• type StatusHelper
  • func NewStatusHelper(cli *clientv3.Client) *StatusHelper
  • func (sh *StatusHelper) ReadStatus(id string) (*config.Status, error)
  • func (sh *StatusHelper) WriteStatus(id string, status string, info string) error
• type TaskQueue
  • func NewTaskQueue() TaskQueue
  • func (q *TaskQueue) Empty() bool
  • func (q *TaskQueue) Put(element *Service)
  • func (q *TaskQueue) RemoveByID(id string) bool
  • func (q *TaskQueue) Take() (*Service, bool)

Constants

const (
	StatusPrefix = "/hera/dm/status/"

	TipStatusPending        = "Join in queue, wait for execute. My WindowPeriod is %d - %d "
	TipStatusRunning        = "I M Running ~"
	TipStatusError          = "Error Occurred : %s ."
	TipStatusFinished       = "Job Finished at %s "
	TipStatusStoppedRunning = "Manually stopped when running"
	TipStatusStoppedPending = "Manually stopped when still pending"

	StatusPendingError    = "Status is pending but could not find it in the taskQueue "
	StatusCannotStopError = "Status is %s . This status cannot be stopped "
)

const (
	StopKeyPrefix = "/hera/dm/stop-job/"
)

Variables

var CountOfJob = prometheus2.NewCounterVec(
	prometheus2.CounterOpts{
		Name: "dm_task_total_count",
		Help: "Number of Job",
	},
	[]string{},
)

var CountOfJobError = prometheus2.NewCounterVec(
	prometheus2.CounterOpts{
		Name: "dm_task_error",
		Help: "Number of Job Occurred Error In Dm",
	},
	[]string{},
)

var CountOfJobFinished = prometheus2.NewCounterVec(
	prometheus2.CounterOpts{
		Name: "dm_task_finished",
		Help: "Number of Job Finished By Dm",
	},
	[]string{},
)

var CountOfRunningGoRoutine = prometheus2.NewGauge(
	prometheus2.GaugeOpts{
		Name: "dm_go_routine_count",
		Help: "Number of Go Routine Running",
	},
)

Functions

func DecCountOfRunningGoRoutine

func DecCountOfRunningGoRoutine()

func IncCountOfRunningGoRoutine

func IncCountOfRunningGoRoutine()

func IncCountOfTask

func IncCountOfTask()

func IncCountOfTaskError

func IncCountOfTaskError()

func IncCountOfTaskFinished

func IncCountOfTaskFinished()

func NewStatus

func NewStatus(id string, status string, info string, updateTime time.Time) *config.Status

Types

type DmPrometheus

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

func NewDmPrometheus

func NewDmPrometheus(pc *prometheus.Config, onOff chan bool) *DmPrometheus

func (*DmPrometheus) Run

func (dp *DmPrometheus) Run()

func (*DmPrometheus) Stop

func (dp *DmPrometheus) Stop()

type Manager

type Manager struct {
	Controllers  map[string]*chan bool
	Tasks        *TaskQueue
	Config       *config.Config
	CountManager *chan int
	StatusHelper *StatusHelper
}

func NewManager

func NewManager(controllers map[string]*chan bool, tasks *TaskQueue, config *config.Config, countManager *chan int, statusHelper *StatusHelper) *Manager

func (*Manager) AddController

func (m *Manager) AddController(taskID string) *chan bool

func (*Manager) Consume

func (m *Manager) Consume()

func (*Manager) HandleJob

func (m *Manager) HandleJob(job *config.Job, mutex *concurrency.Mutex) error

func (*Manager) StopJob

func (m *Manager) StopJob(service *Service) error

type Service

type Service struct {
	Job        *config.Job
	Conf       *config.Config
	Controller *chan bool
	IDType     string
	Mutex      *concurrency.Mutex
	StopChan   *chan bool
}

func NewService

func NewService(job *config.Job, config *config.Config, controller *chan bool, idType string, mutex *concurrency.Mutex, stopChan *chan bool) *Service

func (*Service) Run

func (s *Service) Run() (string, error)

type StatusHelper

type StatusHelper struct {
	Client *clientv3.Client
}

func NewStatusHelper

func NewStatusHelper(cli *clientv3.Client) *StatusHelper

func (*StatusHelper) ReadStatus

func (sh *StatusHelper) ReadStatus(id string) (*config.Status, error)

func (*StatusHelper) WriteStatus

func (sh *StatusHelper) WriteStatus(id string, status string, info string) error

type TaskQueue

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

func NewTaskQueue

func NewTaskQueue() TaskQueue

func (*TaskQueue) Empty

func (q *TaskQueue) Empty() bool

func (*TaskQueue) Put

func (q *TaskQueue) Put(element *Service)

func (*TaskQueue) RemoveByID

func (q *TaskQueue) RemoveByID(id string) bool

func (*TaskQueue) Take

func (q *TaskQueue) Take() (*Service, bool)