Daemon subdaemon refactor

core/collector/rpc.go

@@ -37,7 +37,7 @@ func (c Client) NewPinger(d time.Duration) func() {
 		for {
 			select {
 			case <-stop:
-				break
+				return
 			case <-ticker.C:
 				c.Ping()
 			}

core/commands/daemon_restart.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/opensvc/om3/core/client"
 	"github.com/opensvc/om3/core/nodeselector"
-	"github.com/opensvc/om3/daemon/daemoncli"
+	"github.com/opensvc/om3/daemon/daemoncmd"
 )
 
 type (
@@ -43,7 +43,7 @@ func (t *CmdDaemonRestart) restartLocal() error {
 		return err
 	}
 	ctx := context.Background()
-	return daemoncli.NewContext(ctx, cli).RestartFromCmd(ctx)
+	return daemoncmd.NewContext(ctx, cli).RestartFromCmd(ctx)
 }
 
 func (t *CmdDaemonRestart) restartRemotes(nodes []string) error {

core/commands/daemon_running.go

@@ -3,7 +3,7 @@ package commands
 import (
 	"os"
 
-	"github.com/opensvc/om3/daemon/daemoncli"
+	"github.com/opensvc/om3/daemon/daemoncmd"
 )
 
 type (
@@ -17,7 +17,7 @@ func (t *CmdDaemonRunning) Run() error {
 	if err != nil {
 		return err
 	}
-	dCli := daemoncli.New(cli)
+	dCli := daemoncmd.New(cli)
 	dCli.SetNode(t.NodeSelector)
 	if !dCli.Running() {
 		os.Exit(1)

core/commands/daemon_start.go

@@ -3,7 +3,7 @@ package commands
 import (
 	"context"
 
-	"github.com/opensvc/om3/daemon/daemoncli"
+	"github.com/opensvc/om3/daemon/daemoncmd"
 )
 
 type (
@@ -32,5 +32,5 @@ func (t *CmdDaemonStart) Run() error {
 		return err
 	}
 	ctx := context.Background()
-	return daemoncli.NewContext(ctx, cli).StartFromCmd(ctx, t.Foreground, t.CpuProfile)
+	return daemoncmd.NewContext(ctx, cli).StartFromCmd(ctx, t.Foreground, t.CpuProfile)
 }

core/commands/daemon_stop.go

@@ -3,7 +3,7 @@ package commands
 import (
 	"context"
 
-	"github.com/opensvc/om3/daemon/daemoncli"
+	"github.com/opensvc/om3/daemon/daemoncmd"
 )
 
 type (
@@ -18,5 +18,5 @@ func (t *CmdDaemonStop) Run() error {
 		return err
 	}
 	ctx := context.Background()
-	return daemoncli.NewContext(ctx, cli).StopFromCmd(ctx)
+	return daemoncmd.NewContext(ctx, cli).StopFromCmd(ctx)
 }

daemon/ccfg/main.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -21,7 +22,6 @@ import (
 	"github.com/opensvc/om3/core/node"
 	"github.com/opensvc/om3/core/object"
 	"github.com/opensvc/om3/core/xconfig"
-	"github.com/opensvc/om3/daemon/daemondata"
 	"github.com/opensvc/om3/daemon/draincommand"
 	"github.com/opensvc/om3/daemon/msgbus"
 	"github.com/opensvc/om3/util/hostname"
@@ -37,7 +37,7 @@ type (
 		ctx           context.Context
 		cancel        context.CancelFunc
 		cmdC          chan any
-		databus       *daemondata.T
+		drainDuration time.Duration
 		bus           *pubsub.Bus
 		log           zerolog.Logger
 		startedAt     time.Time
@@ -53,6 +53,7 @@ type (
 		change      bool
 
 		sub *pubsub.Subscription
+		wg  sync.WaitGroup
 	}
 
 	cmdGet struct {
@@ -68,20 +69,21 @@ var (
 	cmdC chan any
 )
 
-// Start launches the ccfg worker goroutine
-func Start(parent context.Context, drainDuration time.Duration) error {
-	ctx, cancel := context.WithCancel(parent)
-
+func New(drainDuration time.Duration) *ccfg {
 	o := &ccfg{
-		networkSigs: make(map[string]string),
-		ctx:         ctx,
-		cancel:      cancel,
-		cmdC:        make(chan any),
-		databus:     daemondata.FromContext(ctx),
-		bus:         pubsub.BusFromContext(ctx),
-		log:         log.Logger.With().Str("func", "ccfg").Logger(),
-		localhost:   hostname.Hostname(),
+		cmdC:          make(chan any),
+		drainDuration: drainDuration,
+		localhost:     hostname.Hostname(),
+		log:           log.Logger.With().Str("func", "ccfg").Logger(),
+		networkSigs:   make(map[string]string),
 	}
+	return o
+}
+
+// Start launches the ccfg worker goroutine
+func (o *ccfg) Start(parent context.Context) error {
+	o.ctx, o.cancel = context.WithCancel(parent)
+	o.bus = pubsub.BusFromContext(o.ctx)
 	cmdC = o.cmdC
 
 	if n, err := object.NewCluster(object.WithVolatile(true)); err != nil {
@@ -93,12 +95,14 @@ func Start(parent context.Context, drainDuration time.Duration) error {
 	o.pubClusterConfig()
 
 	o.startSubscriptions()
+	o.wg.Add(1)
 	go func() {
 		defer func() {
-			draincommand.Do(o.cmdC, drainDuration)
+			draincommand.Do(o.cmdC, o.drainDuration)
 			if err := o.sub.Stop(); err != nil && !errors.Is(err, context.Canceled) {
 				o.log.Warn().Err(err).Msg("subscription stop")
 			}
+			o.wg.Done()
 		}()
 		o.worker()
 	}()
@@ -109,6 +113,12 @@ func Start(parent context.Context, drainDuration time.Duration) error {
 	return nil
 }
 
+func (o *ccfg) Stop() error {
+	o.cancel()
+	o.wg.Wait()
+	return nil
+}
+
 func (o *ccfg) startSubscriptions() {
 	sub := o.bus.Sub("ccfg")
 	sub.AddFilter(&msgbus.ConfigFileUpdated{}, pubsub.Label{"path", "cluster"})

daemon/cstat/main.go

@@ -8,14 +8,14 @@ package cstat
 import (
 	"context"
 	"errors"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
 	"github.com/opensvc/om3/core/cluster"
 	"github.com/opensvc/om3/core/node"
-	"github.com/opensvc/om3/daemon/daemondata"
 	"github.com/opensvc/om3/daemon/msgbus"
 	"github.com/opensvc/om3/util/pubsub"
 )
@@ -27,7 +27,6 @@ type (
 		ctx       context.Context
 		cancel    context.CancelFunc
 		cmdC      chan any
-		databus   *daemondata.T
 		bus       *pubsub.Bus
 		log       zerolog.Logger
 		startedAt time.Time
@@ -41,26 +40,28 @@ type (
 		change      bool
 
 		sub *pubsub.Subscription
+		wg  sync.WaitGroup
 	}
 )
 
-// Start launches the cstat worker goroutine
-func Start(parent context.Context) error {
-	ctx, cancel := context.WithCancel(parent)
-
-	o := &cstat{
-		ctx:        ctx,
-		cancel:     cancel,
-		databus:    daemondata.FromContext(ctx),
-		bus:        pubsub.BusFromContext(ctx),
+func New() *cstat {
+	return &cstat{
 		log:        log.Logger.With().Str("func", "cstat").Logger(),
 		nodeStatus: make(map[string]node.Status),
 	}
+}
+
+// Start launches the cstat worker goroutine
+func (o *cstat) Start(parent context.Context) error {
+	o.ctx, o.cancel = context.WithCancel(parent)
+	o.bus = pubsub.BusFromContext(o.ctx)
 
 	o.startSubscriptions()
+	o.wg.Add(1)
 	go func() {
+		defer o.wg.Done()
 		defer func() {
-			if err := o.sub.Stop(); err != nil && !errors.Is(err, context.Canceled){
+			if err := o.sub.Stop(); err != nil && !errors.Is(err, context.Canceled) {
 				o.log.Error().Err(err).Msg("subscription stop")
 			}
 		}()
@@ -69,6 +70,12 @@ func Start(parent context.Context) error {
 	return nil
 }
 
+func (o *cstat) Stop() error {
+	o.cancel()
+	o.wg.Wait()
+	return nil
+}
+
 func (o *cstat) startSubscriptions() {
 	sub := o.bus.Sub("cstat")
 	sub.AddFilter(&msgbus.NodeStatusUpdated{})