ctrl/processes.go

package ctrl

import (
	"context"
	"fmt"
	"log"
	"sync"
)

// processes holds all the information about running processes
type processes struct {
	// mutex for processes
	mu sync.Mutex
	// The main context for subscriber processes.
	ctx context.Context
	// cancel func to send cancel signal to the subscriber processes context.
	cancel context.CancelFunc
	// The active spawned processes
	// server
	server *server
	active procsMap
	// mutex to lock the map
	// mu sync.RWMutex
	// The last processID created
	lastProcessID int
	// The instance global prometheus registry.
	metrics *metrics
	// Waitgroup to keep track of all the processes started.
	wg sync.WaitGroup
	// errorKernel
	errorKernel *errorKernel
	// configuration
	configuration *Configuration

	// Signatures
	nodeAuth *nodeAuth
}

// newProcesses will prepare and return a *processes which
// is map containing all the currently running processes.
func newProcesses(ctx context.Context, server *server) *processes {
	p := processes{
		server:        server,
		active:        *newProcsMap(),
		errorKernel:   server.errorKernel,
		configuration: server.configuration,
		nodeAuth:      server.nodeAuth,
		metrics:       server.metrics,
	}

	// Prepare the parent context for the subscribers.
	ctx, cancel := context.WithCancel(ctx)

	// // Start the processes map.
	// go func() {
	// 	p.active.run(ctx)
	// }()

	p.ctx = ctx
	p.cancel = cancel

	return &p
}

// ----------------------

// ----------------------

type procsMap struct {
	procNames map[processName]process
	mu        sync.Mutex
}

func newProcsMap() *procsMap {
	cM := procsMap{
		procNames: make(map[processName]process),
	}
	return &cM
}

// ----------------------

// Start all the subscriber processes.
// Takes an initial process as it's input. All processes
// will be tied to this single process's context.
func (p *processes) Start(proc process) {
	// Set the context for the initial process.
	proc.ctx = p.ctx

	// --- Subscriber services that can be started via flags

	proc.startup.startProcess(proc, OpProcessList, nil)
	proc.startup.startProcess(proc, OpProcessStart, nil)
	proc.startup.startProcess(proc, OpProcessStop, nil)
	proc.startup.startProcess(proc, Test, nil)

	if proc.configuration.StartProcesses.StartSubFileAppend {
		proc.startup.startProcess(proc, FileAppend, nil)
	}

	if proc.configuration.StartProcesses.StartSubFile {
		proc.startup.startProcess(proc, File, nil)
	}

	if proc.configuration.StartProcesses.StartSubCopySrc {
		proc.startup.startProcess(proc, CopySrc, nil)
	}

	if proc.configuration.StartProcesses.StartSubCopyDst {
		proc.startup.startProcess(proc, CopyDst, nil)
	}

	if proc.configuration.StartProcesses.StartSubHello {

		proc.startup.startProcess(proc, Hello, procFuncHello)
	}

	if proc.configuration.StartProcesses.IsCentralErrorLogger {
		proc.startup.startProcess(proc, ErrorLog, nil)
	}

	if proc.configuration.StartProcesses.StartSubCliCommand {
		proc.startup.startProcess(proc, CliCommand, nil)
	}

	if proc.configuration.StartProcesses.StartSubConsole {
		proc.startup.startProcess(proc, Console, nil)
	}

	if proc.configuration.StartProcesses.StartPubHello != 0 {
		proc.startup.startProcess(proc, HelloPublisher, procFuncHelloPublisher)
	}

	if proc.configuration.StartProcesses.EnableKeyUpdates {
		proc.startup.startProcess(proc, KeysRequestUpdate, procFuncKeysRequestUpdate)
		proc.startup.startProcess(proc, KeysDeliverUpdate, nil)
	}

	if proc.configuration.StartProcesses.EnableAclUpdates {

		proc.startup.startProcess(proc, AclRequestUpdate, procFuncAclRequestUpdate)
		proc.startup.startProcess(proc, AclDeliverUpdate, nil)
	}

	if proc.configuration.StartProcesses.IsCentralAuth {
		proc.startup.startProcess(proc, KeysRequestUpdate, nil)
		proc.startup.startProcess(proc, KeysAllow, nil)
		proc.startup.startProcess(proc, KeysDelete, nil)
		proc.startup.startProcess(proc, AclRequestUpdate, nil)
		proc.startup.startProcess(proc, AclAddCommand, nil)
		proc.startup.startProcess(proc, AclDeleteCommand, nil)
		proc.startup.startProcess(proc, AclDeleteSource, nil)
		proc.startup.startProcess(proc, AclGroupNodesAddNode, nil)
		proc.startup.startProcess(proc, AclGroupNodesDeleteNode, nil)
		proc.startup.startProcess(proc, AclGroupNodesDeleteGroup, nil)
		proc.startup.startProcess(proc, AclGroupCommandsAddCommand, nil)
		proc.startup.startProcess(proc, AclGroupCommandsDeleteCommand, nil)
		proc.startup.startProcess(proc, AclGroupCommandsDeleteGroup, nil)
		proc.startup.startProcess(proc, AclExport, nil)
		proc.startup.startProcess(proc, AclImport, nil)
	}

	if proc.configuration.StartProcesses.StartSubHttpGet {
		proc.startup.startProcess(proc, HttpGet, nil)
	}

	if proc.configuration.StartProcesses.StartSubTailFile {
		proc.startup.startProcess(proc, TailFile, nil)
	}

	if proc.configuration.StartProcesses.StartSubCliCommandCont {
		proc.startup.startProcess(proc, CliCommandCont, nil)
	}

	proc.startup.startProcess(proc, PublicKey, nil)
}

// Stop all subscriber processes.
func (p *processes) Stop() {
	log.Printf("info: canceling all subscriber processes...\n")
	p.cancel()
	p.wg.Wait()
	log.Printf("info: done canceling all subscriber processes.\n")

}

// ---------------------------------------------------------------------------------------

// Startup holds all the startup methods for subscribers.
type startup struct {
	server      *server
	centralAuth *centralAuth
	metrics     *metrics
}

func newStartup(server *server) *startup {
	s := startup{
		server:      server,
		centralAuth: server.centralAuth,
		metrics:     server.metrics,
	}

	return &s
}

// startProcess will start a process. It takes the initial process, request method,
// and a procFunc as it's input arguments. If a procFunc is not needed, use the value nil.
func (s *startup) startProcess(p process, m Method, pf func(ctx context.Context, proc process, procFuncCh chan Message) error) {
	er := fmt.Errorf("starting %v subscriber: %#v", m, p.node)
	p.errorKernel.logDebug(er)

	var sub Subject
	switch {
	case m == ErrorLog:
		sub = newSubject(m, "errorCentral")
	default:
		sub = newSubject(m, string(p.node))
	}

	fmt.Printf("DEBUG:::startup subscriber, subject: %v\n", sub)
	proc := newProcess(p.ctx, p.processes.server, sub)
	proc.procFunc = pf

	go proc.start()
}

// ---------------------------------------------------------------

// Print the content of the processes map.
func (p *processes) printProcessesMap() {
	er := fmt.Errorf("output of processes map : ")
	p.errorKernel.logDebug(er)

	{
		p.active.mu.Lock()

		for pName, proc := range p.active.procNames {
			er := fmt.Errorf("info: proc - procName in map: %v , id: %v, subject: %v", pName, proc.processID, proc.subject.name())
			proc.errorKernel.logDebug(er)
		}

		p.metrics.promProcessesTotal.Set(float64(len(p.active.procNames)))

		p.active.mu.Unlock()
	}

}