netdata/netdata

// SPDX-License-Identifier: GPL-3.0-or-later

package jobmgr

import (
    "context"
    "fmt"
    "io"
    "log/slog"
    "os"
    "sync"
    "time"

    "github.com/netdata/netdata/go/go.d.plugin/agent/confgroup"
    "github.com/netdata/netdata/go/go.d.plugin/agent/functions"
    "github.com/netdata/netdata/go/go.d.plugin/agent/module"
    "github.com/netdata/netdata/go/go.d.plugin/agent/netdataapi"
    "github.com/netdata/netdata/go/go.d.plugin/agent/safewriter"
    "github.com/netdata/netdata/go/go.d.plugin/agent/ticker"
    "github.com/netdata/netdata/go/go.d.plugin/logger"

    "github.com/mattn/go-isatty"
    "gopkg.in/yaml.v2"
)

var isTerminal = isatty.IsTerminal(os.Stdout.Fd()) || isatty.IsTerminal(os.Stdin.Fd())

func New() *Manager {
    mgr := &Manager{
        Logger: logger.New().With(
            slog.String("component", "job manager"),
        ),
        Out:             io.Discard,
        FileLock:        noop{},
        FileStatus:      noop{},
        FileStatusStore: noop{},
        Vnodes:          noop{},
        FnReg:           noop{},

        discoveredConfigs: newDiscoveredConfigsCache(),
        seenConfigs:       newSeenConfigCache(),
        exposedConfigs:    newExposedConfigCache(),
        runningJobs:       newRunningJobsCache(),
        retryingTasks:     newRetryingTasksCache(),

        started:  make(chan struct{}),
        api:      netdataapi.New(safewriter.Stdout),
        addCh:    make(chan confgroup.Config),
        rmCh:     make(chan confgroup.Config),
        dyncfgCh: make(chan functions.Function),
    }

    return mgr
}

type Manager struct {
    *logger.Logger

    PluginName     string
    Out            io.Writer
    Modules        module.Registry
    ConfigDefaults confgroup.Registry

    FileLock        FileLocker
    FileStatus      FileStatus
    FileStatusStore FileStatusStore
    Vnodes          Vnodes
    FnReg           FunctionRegistry

    discoveredConfigs *discoveredConfigs
    seenConfigs       *seenConfigs
    exposedConfigs    *exposedConfigs
    retryingTasks     *retryingTasks
    runningJobs       *runningJobs

    ctx      context.Context
    started  chan struct{}
    api      dyncfgAPI
    addCh    chan confgroup.Config
    rmCh     chan confgroup.Config
    dyncfgCh chan functions.Function

    waitCfgOnOff string // block processing of discovered configs until "enable"/"disable" is received from Netdata
}

func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
    m.Info("instance is started")
    defer func() { m.cleanup(); m.Info("instance is stopped") }()
    m.ctx = ctx

    m.FnReg.Register("config", m.dyncfgConfig)

    for name := range m.Modules {
        m.dyncfgModuleCreate(name)
    }

    var wg sync.WaitGroup

    wg.Add(1)
    go func() { defer wg.Done(); m.runProcessConfGroups(in) }()

    wg.Add(1)
    go func() { defer wg.Done(); m.run() }()

    wg.Add(1)
    go func() { defer wg.Done(); m.runNotifyRunningJobs() }()

    close(m.started)

    wg.Wait()
    <-m.ctx.Done()
}

func (m *Manager) runProcessConfGroups(in chan []*confgroup.Group) {
    for {
        select {
        case <-m.ctx.Done():
            return
        case groups := <-in:
            for _, gr := range groups {
                a, r := m.discoveredConfigs.add(gr)
                m.Debugf("received configs: %d/+%d/-%d ('%s')", len(gr.Configs), len(a), len(r), gr.Source)
                sendConfigs(m.ctx, m.rmCh, r...)
                sendConfigs(m.ctx, m.addCh, a...)
            }
        }
    }
}

func (m *Manager) run() {
    for {
        if m.waitCfgOnOff != "" {
            select {
            case <-m.ctx.Done():
                return
            case fn := <-m.dyncfgCh:
                m.dyncfgConfigExec(fn)
            }
        } else {
            select {
            case <-m.ctx.Done():
                return
            case cfg := <-m.addCh:
                m.addConfig(cfg)
            case cfg := <-m.rmCh:
                m.removeConfig(cfg)
            case fn := <-m.dyncfgCh:
                m.dyncfgConfigExec(fn)
            }
        }
    }
}

func (m *Manager) addConfig(cfg confgroup.Config) {
    if _, ok := m.Modules.Lookup(cfg.Module()); !ok {
        return
    }

    m.retryingTasks.remove(cfg)

    scfg, ok := m.seenConfigs.lookup(cfg)
    if !ok {
        scfg = &seenConfig{cfg: cfg}
        m.seenConfigs.add(scfg)
    }

    ecfg, ok := m.exposedConfigs.lookup(cfg)
    if !ok {
        scfg.status = dyncfgAccepted
        ecfg = scfg
        m.exposedConfigs.add(ecfg)
    } else {
        sp, ep := scfg.cfg.SourceTypePriority(), ecfg.cfg.SourceTypePriority()
        if ep > sp || (ep == sp && ecfg.status == dyncfgRunning) {
            m.retryingTasks.remove(cfg)
            return
        }
        if ecfg.status == dyncfgRunning {
            m.stopRunningJob(ecfg.cfg.FullName())
            m.FileLock.Unlock(ecfg.cfg.FullName())
            m.FileStatus.Remove(ecfg.cfg)
        }
        scfg.status = dyncfgAccepted
        m.exposedConfigs.add(scfg) // replace existing exposed
        ecfg = scfg
    }

    m.dyncfgJobCreate(ecfg.cfg, ecfg.status)

    if isTerminal || m.PluginName == "nodyncfg" { // FIXME: quick fix of TestAgent_Run (agent_test.go)
        m.dyncfgConfigEnable(functions.Function{Args: []string{dyncfgJobID(ecfg.cfg), "enable"}})
    } else {
        m.waitCfgOnOff = ecfg.cfg.FullName()
    }
}

func (m *Manager) removeConfig(cfg confgroup.Config) {
    m.retryingTasks.remove(cfg)

    scfg, ok := m.seenConfigs.lookup(cfg)
    if !ok {
        return
    }
    m.seenConfigs.remove(cfg)

    ecfg, ok := m.exposedConfigs.lookup(cfg)
    if !ok || scfg.cfg.UID() != ecfg.cfg.UID() {
        return
    }

    m.exposedConfigs.remove(cfg)
    m.stopRunningJob(cfg.FullName())
    m.FileLock.Unlock(cfg.FullName())
    m.FileStatus.Remove(cfg)

    if !isStock(cfg) || ecfg.status == dyncfgRunning {
        m.dyncfgJobRemove(cfg)
    }
}

func (m *Manager) runNotifyRunningJobs() {
    tk := ticker.New(time.Second)
    defer tk.Stop()

    for {
        select {
        case <-m.ctx.Done():
            return
        case clock := <-tk.C:
            m.runningJobs.lock()
            m.runningJobs.forEach(func(_ string, job *module.Job) { job.Tick(clock) })
            m.runningJobs.unlock()
        }
    }
}

func (m *Manager) startRunningJob(job *module.Job) {
    m.runningJobs.lock()
    defer m.runningJobs.unlock()

    if job, ok := m.runningJobs.lookup(job.FullName()); ok {
        job.Stop()
    }

    go job.Start()
    m.runningJobs.add(job.FullName(), job)
}

func (m *Manager) stopRunningJob(name string) {
    m.runningJobs.lock()
    defer m.runningJobs.unlock()

    if job, ok := m.runningJobs.lookup(name); ok {
        job.Stop()
        m.runningJobs.remove(name)
    }
}

func (m *Manager) cleanup() {
    m.FnReg.Unregister("config")

    m.runningJobs.lock()
    defer m.runningJobs.unlock()

    m.runningJobs.forEach(func(key string, job *module.Job) {
        job.Stop()
    })
}

func (m *Manager) createCollectorJob(cfg confgroup.Config) (*module.Job, error) {
    creator, ok := m.Modules[cfg.Module()]
    if !ok {
        return nil, fmt.Errorf("can not find %s module", cfg.Module())
    }

    var vnode struct {
        guid     string
        hostname string
        labels   map[string]string
    }

    if cfg.Vnode() != "" {
        n, ok := m.Vnodes.Lookup(cfg.Vnode())
        if !ok {
            return nil, fmt.Errorf("vnode '%s' is not found", cfg.Vnode())
        }

        vnode.guid = n.GUID
        vnode.hostname = n.Hostname
        vnode.labels = n.Labels
    }

    m.Debugf("creating %s[%s] job, config: %v", cfg.Module(), cfg.Name(), cfg)

    mod := creator.Create()

    if err := applyConfig(cfg, mod); err != nil {
        return nil, err
    }

    jobCfg := module.JobConfig{
        PluginName:      m.PluginName,
        Name:            cfg.Name(),
        ModuleName:      cfg.Module(),
        FullName:        cfg.FullName(),
        UpdateEvery:     cfg.UpdateEvery(),
        AutoDetectEvery: cfg.AutoDetectionRetry(),
        Priority:        cfg.Priority(),
        Labels:          makeLabels(cfg),
        IsStock:         cfg.SourceType() == "stock",
        Module:          mod,
        Out:             m.Out,
        VnodeGUID:       vnode.guid,
        VnodeHostname:   vnode.hostname,
        VnodeLabels:     vnode.labels,
    }

    job := module.NewJob(jobCfg)

    return job, nil
}

func runRetryTask(ctx context.Context, out chan<- confgroup.Config, cfg confgroup.Config) {
    t := time.NewTimer(time.Second * time.Duration(cfg.AutoDetectionRetry()))
    defer t.Stop()

    select {
    case <-ctx.Done():
    case <-t.C:
        sendConfigs(ctx, out, cfg)
    }
}

func sendConfigs(ctx context.Context, out chan<- confgroup.Config, cfgs ...confgroup.Config) {
    for _, cfg := range cfgs {
        select {
        case <-ctx.Done():
            return
        case out <- cfg:
        }
    }
}

func isStock(cfg confgroup.Config) bool {
    return cfg.SourceType() == confgroup.TypeStock
}

func isDyncfg(cfg confgroup.Config) bool {
    return cfg.SourceType() == confgroup.TypeDyncfg
}

func applyConfig(cfg confgroup.Config, module any) error {
    bs, err := yaml.Marshal(cfg)
    if err != nil {
        return err
    }
    return yaml.Unmarshal(bs, module)
}

func makeLabels(cfg confgroup.Config) map[string]string {
    labels := make(map[string]string)
    for name, value := range cfg.Labels() {
        n, ok1 := name.(string)
        v, ok2 := value.(string)
        if ok1 && ok2 {
            labels[n] = v
        }
    }
    return labels
}