module

Documentation

Index

• Constants
• Variables
• func DecCountOfRunningGoRoutine()
• func IncCountOfRunningGoRoutine()
• func IncCountOfTask()
• func IncCountOfTaskError()
• func IncCountOfTaskFinished()
• func NewStatus(id, status, 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, status, 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, status, 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, status, 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)