$ tree nightwatch
nightwatch # 項(xiàng)目目錄
├── README.md # README 文件
├── assets # 項(xiàng)目相關(guān)的資源目錄
│   ├── docker-compose.yaml # 用于啟動(dòng)項(xiàng)目依賴(lài)的 MariaDB 和 Redis
│   └── schema.sql # 測(cè)試數(shù)據(jù) SQL
├── cmd # 項(xiàng)目啟動(dòng)入口
│   └── main.go
├── go.mod
├── go.sum
├── internal # 項(xiàng)目?jī)?nèi)部包
│   ├── logger.go # 定制日志
│   ├── nightwatch.go # nightwatch 的實(shí)現(xiàn)和啟動(dòng)入口
│   └── watcher # 任務(wù)接口和實(shí)現(xiàn)
│       ├── all # 任務(wù)注冊(cè)入口
│       │   └── all.go
│       ├── config.go # 任務(wù)配置
│       ├── task # 任務(wù)實(shí)現(xiàn)，一個(gè)可以定時(shí)同步 MariaDB 和 Kubernetes 任務(wù)狀態(tài)的示例程序
│       │   ├── task.go
│       │   └── watcher.go
│       └── watcher.go # 任務(wù)接口
└── pkg # 項(xiàng)目公共包
    ├── db # 數(shù)據(jù)庫(kù)實(shí)例
    │   ├── mysql.go
    │   └── redis.go
    ├── meta # 元信息
    │   └── where.go # MariaDB where 查詢(xún)條件元信息封裝
    ├── model # 任務(wù)模型
    │   └── task.go
    ├── store # 數(shù)據(jù)庫(kù)操作接口
    │   ├── helper.go
    │   ├── store.go
    │   └── task.go
    └── util # 工具包
        └── reflect
            └── reflect.go

14 directories, 21 files

CREATE TABLE IF NOTEXISTS `task` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `name` varchar(45) NOT NULLDEFAULT'' COMMENT '任務(wù)名稱(chēng)',
  `namespace` varchar(45) NOT NULLDEFAULT'' COMMENT 'k8s namespace 名稱(chēng)',
  `info` TEXT NOT NULL COMMENT '任務(wù) k8s 相關(guān)信息',
  `status` varchar(45) NOT NULLDEFAULT'' COMMENT '任務(wù)狀態(tài)',
  `user_id` bigint(20) NOT NULLDEFAULT'0' COMMENT '用戶(hù) ID',
  `created_at` datetime NOT NULLDEFAULTCURRENT_TIMESTAMP,
  `updated_at` datetime NOT NULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_name_namespace` (`name`, `namespace`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='任務(wù)表';

INSERT INTO `task` (`id`, `name`, `namespace`, `info`, `status`, `user_id`) VALUES (1, 'demo-task-1', 'default', '{"image":"alpine","command":["sleep"],"args":["60"]}', 'Normal', 1);
INSERT INTO `task` (`id`, `name`, `namespace`, `info`, `status`, `user_id`) VALUES (2, 'demo-task-2', 'demo', '{"image":"busybox","command":["sleep"],"args":["3600"]}', 'Normal', 2);

package main

import (
    "flag"
    "log/slog"
    "path/filepath"
    "time"

    genericapiserver "k8s.io/apiserver/pkg/server"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/homedir"

    "github.com/jianghushinian/blog-go-example/nightwatch/internal"
    "github.com/jianghushinian/blog-go-example/nightwatch/pkg/db"
)

func main() {
    slog.SetLogLoggerLevel(slog.LevelDebug)

    var kubecfg *string
    if home := homedir.HomeDir(); home != "" {
        kubecfg = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "Optional absolute path to kubeconfig")
    } else {
        kubecfg = flag.String("kubeconfig", "", "Absolute path to kubeconfig")
    }

    config, err := clientcmd.BuildConfigFromFlags("", *kubecfg)
    if err != nil {
        slog.Error(err.Error())
        return
    }
    config.QPS = 50
    config.Burst = 100
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        slog.Error(err.Error())
        return
    }

    cfg := nightwatch.Config{
        MySQLOptions: &db.MySQLOptions{
            Host:                  "127.0.0.1:33306",
            Username:              "root",
            Password:              "nightwatch",
            Database:              "nightwatch",
            MaxIdleConnections:    100,
            MaxOpenConnections:    100,
            MaxConnectionLifeTime: time.Duration(10) * time.Second,
        },
        RedisOptions: &db.RedisOptions{
            Addr:         "127.0.0.1:36379",
            Username:     "",
            Password:     "nightwatch",
            Database:     0,
            MaxRetries:   3,
            MinIdleConns: 0,
            DialTimeout:  5 * time.Second,
            ReadTimeout:  3 * time.Second,
            WriteTimeout: 3 * time.Second,
            PoolSize:     10,
        },
        Clientset: clientset,
    }

    nw, err := cfg.New()
    if err != nil {
        slog.Error(err.Error())
        return
    }

    stopCh := genericapiserver.SetupSignalHandler()
    nw.Run(stopCh)
}

// New 通過(guò)配置構(gòu)造一個(gè) nightWatch 對(duì)象
func (c *Config) New() (*nightWatch, error) {
    rdb, err := db.NewRedis(c.RedisOptions)
    if err != nil {
        slog.Error(err.Error(), "Failed to create Redis client")
        returnnil, err
    }

    logger := newCronLogger()
    runner := cron.New(
        cron.WithSeconds(),
        cron.WithLogger(logger),
        cron.WithChain(cron.SkipIfStillRunning(logger), cron.Recover(logger)),
    )

    pool := goredis.NewPool(rdb)
    lockOpts := []redsync.Option{
        redsync.WithRetryDelay(50 * time.Microsecond),
        redsync.WithTries(3),
        redsync.WithExpiry(defaultExpiration),
    }
    locker := redsync.New(pool).NewMutex(lockName, lockOpts...)

    cfg, err := c.CreateWatcherConfig()
    if err != nil {
        returnnil, err
    }

    nw := &nightWatch{runner: runner, locker: locker, config: cfg}
    if err := nw.addWatchers(); err != nil {
        returnnil, err
    }

    return nw, nil
}

// 注冊(cè)所有 Watcher 實(shí)例到 nightWatch
func (nw *nightWatch) addWatchers() error {
    for n, w := range watcher.ListWatchers() {
        if err := w.Init(context.Background(), nw.config); err != nil {
            slog.Error(err.Error(), "Failed to construct watcher", "watcher", n)
            return err
        }

        spec := watcher.Every3Seconds
        if obj, ok := w.(watcher.ISpec); ok {
            spec = obj.Spec()
        }

        if _, err := nw.runner.AddJob(spec, w); err != nil {
            slog.Error(err.Error(), "Failed to add job to the cron", "watcher", n)
            return err
        }
    }

    returnnil
}

type Watcher interface {
    Init(ctx context.Context, config *Config) error
    cron.Job
}

type ISpec interface {
    Spec() string
}

var (
    registryLock = new(sync.Mutex)
    registry     = make(map[string]Watcher)
)

func Register(watcher Watcher) {
    registryLock.Lock()
    defer registryLock.Unlock()

    name := reflectutil.StructName(watcher)
    if _, ok := registry[name]; ok {
        panic("duplicate watcher entry: " + name)
    }

    registry[name] = watcher
}

func ListWatchers()map[string]Watcher {
    registryLock.Lock()
    defer registryLock.Unlock()

    return registry
}

type Job interface {
    Run()
}

// Run 執(zhí)行異步任務(wù)，此方法會(huì)阻塞直到關(guān)閉 stopCh
func (nw *nightWatch) Run(stopCh <-chanstruct{}) {
    ctx := wait.ContextForChannel(stopCh)

    // 循環(huán)加鎖，直到加鎖成功，再去啟動(dòng)任務(wù)
    ticker := time.NewTicker(defaultExpiration + (5 * time.Second))
    defer ticker.Stop()
    for {
        err := nw.locker.LockContext(ctx)
        if err == nil {
            slog.Debug("Successfully acquired lock", "lockName", lockName)
            break
        }
        slog.Debug("Failed to acquire lock", "lockName", lockName, "err", err)
        <-ticker.C
    }

    // 看門(mén)狗，實(shí)現(xiàn)鎖自動(dòng)續(xù)約
    ticker = time.NewTicker(extendExpiration)
    defer ticker.Stop()
    gofunc() {
        for {
            select {
            case <-ticker.C:
                if ok, err := nw.locker.ExtendContext(ctx); !ok || err != nil {
                    slog.Debug("Failed to extend lock", "err", err, "status", ok)
                }
            case <-ctx.Done():
                slog.Debug("Exiting lock watchdog")
                return
            }
        }
    }()

    // 啟動(dòng)定時(shí)任務(wù)
    nw.runner.Start()
    slog.Info("Successfully started nightwatch server")

    // 阻塞等待退出信號(hào)
    <-stopCh

    nw.stop()
}

var _ watcher.Watcher = (*taskWatcher)(nil)

type taskWatcher struct {
    store     store.IStore
    clientset kubernetes.Interface

    wg sync.WaitGroup
}

func (w *taskWatcher) Init(ctx context.Context, config *watcher.Config) error {
    w.store = config.Store
    w.clientset = config.Clientset
    returnnil
}

func (w *taskWatcher) Spec() string {
    return"@every 30s"
}

func init() {
    watcher.Register(&taskWatcher{})
}

package all

import (
    // 觸發(fā)所有 Watcher 的 init 函數(shù)進(jìn)行注冊(cè)
    _ "github.com/jianghushinian/blog-go-example/nightwatch/internal/watcher/task"
)

package nightwatch

import (
    ...
    // 觸發(fā) init 函數(shù)
    _ "github.com/jianghushinian/blog-go-example/nightwatch/internal/watcher/all"
)

// Run 運(yùn)行 task watcher 任務(wù)
func (w *taskWatcher) Run() {
    w.wg.Add(2)

    slog.Debug("Current sync period is start")

    // NOTE: 將 Normal 狀態(tài)任務(wù)在 Kubernetes 中啟動(dòng)
    gofunc() {
        defer w.wg.Done()
        ctx := context.Background()

        _, tasks, err := w.store.Tasks().List(ctx, meta.WithFilter(map[string]any{
            "status": model.TaskStatusNormal,
        }))
        if err != nil {
            slog.Error(err.Error(), "Failed to list tasks")
            return
        }

        var wg sync.WaitGroup
        wg.Add(len(tasks))
        for _, task := range tasks {
            gofunc(task *model.Task) {
                defer wg.Done()
                job, err := w.clientset.BatchV1().Jobs(task.Namespace).Create(ctx, toJob(task), metav1.CreateOptions{})
                if err != nil {
                    slog.Error(err.Error(), "Failed to create job")
                    return
                }

                task.Status = model.TaskStatusPending
                if err := w.store.Tasks().Update(ctx, task); err != nil {
                    slog.Error(err.Error(), "Failed to update task status")
                    return
                }
                slog.Info("Successfully created job", "namespace", job.Namespace, "name", job.Name)
            }(task)
        }
        wg.Wait()
    }()

    // NOTE: 同步中間狀態(tài)的任務(wù)在 Kubernetes 中的狀態(tài)到表中
    gofunc() {
        defer w.wg.Done()
        ctx := context.Background()

        _, tasks, err := w.store.Tasks().List(ctx, meta.WithFilterNot(map[string]any{
            // 排除這幾個(gè)狀態(tài)
            "status": []model.TaskStatus{model.TaskStatusNormal, model.TaskStatusSucceeded, model.TaskStatusFailed},
        }))
        if err != nil {
            slog.Error(err.Error(), "Failed to list tasks")
            return
        }

        var wg sync.WaitGroup
        wg.Add(len(tasks))
        for _, task := range tasks {
            gofunc(task *model.Task) {
                defer wg.Done()
                job, err := w.clientset.BatchV1().Jobs(task.Namespace).Get(ctx, task.Name, metav1.GetOptions{})
                if err != nil {
                    slog.Error(err.Error(), "Failed to get task")
                    return
                }

                task.Status = toTaskStatus(job)
                if err := w.store.Tasks().Update(ctx, task); err != nil {
                    slog.Error(err.Error(), "Failed to update task status")
                    return
                }
                slog.Info("Successfully sync job status to task", "namespace", job.Namespace, "name", job.Name, "status", task.Status)
            }(task)
        }
        wg.Wait()
    }()

    w.wg.Wait()
    slog.Debug("Current sync period is complete")
}