lug 源码速览

├── Dockerfile
├── LICENSE
├── README.md
├── cli/lug
│   ├── license.go
│   └── main.go
├── go.mod
├── go.sum
└── pkg
    ├── config
    │   └── config.go
    ├── exporter
    │   └── exporter.go
    ├── helper
    │   ├── disk_usage.go
    │   └── max_length_slice.go
    ├── manager
    │   ├── json_rest.go
    │   └── manager.go
    └── worker
        ├── executor.go
        ├── executor_invoke_worker.go
        ├── external_worker.go
        ├── shell_script_executor.go
        ├── utilities.go
        ├── utility_rlimit.go
        └── worker.go

import (
	"net/http"
	
	"github.com/sjtug/lug/pkg/config"
	"github.com/sjtug/lug/pkg/exporter"
	"github.com/sjtug/lug/pkg/manager"
)
func main() {
	m, err := manager.NewManager(&cfg)
	if err != nil {
		panic(err)
	}
	jsonapi := manager.NewRestfulAPI(m)
	handler := jsonapi.GetAPIHandler()
	go http.ListenAndServe(cfg.JsonAPIConfig.Address, handler)
	go exporter.Expose(cfg.ExporterAddr)
	m.Run()
}

// pkg/manager/json_rest.go
router, err := rest.MakeRouter(
	rest.Get("/lug/v1/admin/manager/detail", r.getManagerStatusDetail),
	rest.Get("/lug/v1/manager/summary", r.getManagerStatusSummary),
	rest.Post("/lug/v1/admin/manager/start", r.startManager),
	rest.Post("/lug/v1/admin/manager/stop", r.stopManager),
	rest.Delete("/lug/v1/admin/manager", r.exitManager),
)

type Manager struct {
	config                *config.Config // main config
	workers               []worker.Worker // sync items
	workersLastInvokeTime map[string]time.Time
	controlChan           chan int // extern control signal
	finishChan            chan int // finish signal
	running               bool
	pendingQueue          []int // storing index of worker to launch
	logger                *logrus.Entry
}

func (m *Manager) Run() {
	c := time.Tick(time.Duration(m.config.Interval) * time.Second)
	for _, w := range m.workers {
		go w.RunSync()
	}
	for {
		// wait until config.Interval seconds has elapsed
		select {
		case <-c:
			if m.running {
				running := 0
				for i, w := range m.workers {
					wStatus := w.GetStatus()
					if !wStatus.Idle {
						running++
						continue
					}
					wConfig := w.GetConfig()
					wTime := m.workersLastInvokeTime[wConfig["name"].(string)]
					elapsed := time.Since(wTime)
					sec2sync, ok := wConfig["interval"].(int)
					if !ok {
						sec2sync = 31536000 // if "interval" is not specified, then worker will launch once a year
					}
					time2sync := time.Duration(sec2sync)*time.Second
					if !m.isAlreadyInPendingQueue(i) && elapsed > time2sync {
						m.pendingQueue = append(m.pendingQueue, i)
					}
				}
				max_allowed := m.config.ConcurrentLimit - running
				new_idx := min(len(m.pendingQueue), max_allowed)
				to_launch := m.pendingQueue[:new_idx]
				m.pendingQueue = m.pendingQueue[new_idx:]
			
				for _, w_idx := range to_launch {
					w := m.workers[w_idx]
					wConfig := w.GetConfig()
					m.workersLastInvokeTime[wConfig["name"].(string)] = time.Now()
					w.TriggerSync()
				}
			}
		case sig, ok := <-m.controlChan:
			if ok {
				switch sig {
				case SigStart:
					m.running = true
					m.finishChan <- StartFinish
				case SigStop:
					m.running = false
					m.finishChan <- StopFinish
				case SigExit:
					goto END_OF_FINISH
				}
			}
		}
	}
END_OF_FINISH:
	m.finishChan <- ExitFinish
}

// Worker declares interface for workers using diffenent ways of sync.
type Worker interface {
	// This call should be thread-safe
	GetStatus() Status
	// This should block forever
	RunSync()
	// This call should be thread-safe
	TriggerSync()

	GetConfig() config.RepoConfig
}