Compare commits
8 Commits
v0.3.3
...
613e21597b
| Author | SHA1 | Date | |
|---|---|---|---|
| 613e21597b | |||
| 126374fbac | |||
| 02f06d857e | |||
| 3d25441e7a | |||
| 735d314a19 | |||
| 4f4c7a0627 | |||
| 461d0b77e9 | |||
| 861d6af1d7 |
@@ -41,6 +41,8 @@ type environment struct {
|
|||||||
logDirectory string
|
logDirectory string
|
||||||
configFile string
|
configFile string
|
||||||
verbose bool
|
verbose bool
|
||||||
|
crash bool
|
||||||
|
crashOnError bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// running stores whether the environment is currently running.
|
// running stores whether the environment is currently running.
|
||||||
@@ -110,7 +112,12 @@ func (this *environment) Add(ctx context.Context, actors ...Actor) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
err := this.initializeActors(ctx, initializable...)
|
err := this.initializeActors(ctx, initializable...)
|
||||||
if err != nil { return err }
|
if err != nil {
|
||||||
|
if this.flags.crashOnError {
|
||||||
|
panic(fmt.Sprint(err))
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
for _, actor := range actors {
|
for _, actor := range actors {
|
||||||
if actor, ok := actor.(Configurable); ok {
|
if actor, ok := actor.(Configurable); ok {
|
||||||
err := actor.Configure(this.conf)
|
err := actor.Configure(this.conf)
|
||||||
@@ -136,11 +143,13 @@ func (this *environment) Del(ctx context.Context, actors ...Actor) error {
|
|||||||
channels := []<- chan struct { } { }
|
channels := []<- chan struct { } { }
|
||||||
for _, actor := range actors {
|
for _, actor := range actors {
|
||||||
info := this.info(actor)
|
info := this.info(actor)
|
||||||
if info.done != nil {
|
if info.stopped != nil {
|
||||||
channels = append(channels, info.stopped)
|
channels = append(channels, info.stopped)
|
||||||
}
|
}
|
||||||
|
info.done()
|
||||||
}
|
}
|
||||||
for _, channel := range channels {
|
for _, channel := range channels {
|
||||||
|
if channel == nil { continue }
|
||||||
select {
|
select {
|
||||||
case <- channel:
|
case <- channel:
|
||||||
case <- ctx.Done():
|
case <- ctx.Done():
|
||||||
@@ -243,13 +252,20 @@ func (this *environment) run(actor Actor) {
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
log.Printf("!!! [%s] stopped with error: %v", typ, stopErr)
|
log.Printf("!!! [%s] stopped with error: %v", typ, stopErr)
|
||||||
|
if this.flags.crashOnError {
|
||||||
|
panic(fmt.Sprint(stopErr))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// contains context information
|
// contains context information
|
||||||
info := this.info(actor)
|
info := this.info(actor)
|
||||||
ctx := info.ctx
|
ctx := info.ctx
|
||||||
defer close(info.stopped)
|
defer func() {
|
||||||
|
if info.stopped != nil {
|
||||||
|
close(info.stopped)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
switch actor := actor.(type) {
|
switch actor := actor.(type) {
|
||||||
case Runnable:
|
case Runnable:
|
||||||
@@ -282,7 +298,12 @@ func (this *environment) runRunnable(ctx context.Context, actor Runnable) (stopE
|
|||||||
for {
|
for {
|
||||||
// run actor
|
// run actor
|
||||||
lastStart := time.Now()
|
lastStart := time.Now()
|
||||||
err := panicWrapCtx(ctx, actor.Run)
|
var err error
|
||||||
|
if this.flags.crash {
|
||||||
|
err = actor.Run(ctx)
|
||||||
|
} else {
|
||||||
|
err = panicWrapCtx(ctx, actor.Run)
|
||||||
|
}
|
||||||
|
|
||||||
// detect context cancellation
|
// detect context cancellation
|
||||||
if ctxErr := ctx.Err(); ctxErr != nil {
|
if ctxErr := ctx.Err(); ctxErr != nil {
|
||||||
@@ -299,11 +320,17 @@ func (this *environment) runRunnable(ctx context.Context, actor Runnable) (stopE
|
|||||||
} else {
|
} else {
|
||||||
// failure
|
// failure
|
||||||
log.Printf("XXX [%s] failed: %v", typ, err)
|
log.Printf("XXX [%s] failed: %v", typ, err)
|
||||||
|
if this.flags.crashOnError {
|
||||||
|
panic(fmt.Sprint(err))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// restart logic
|
// restart logic
|
||||||
if time.Since(lastStart) < restartThreshold {
|
if time.Since(lastStart) < restartThreshold {
|
||||||
log.Printf("!!! [%s] failed too soon, restarting in %v", typ, restartInterval)
|
log.Printf("!!! [%s] failed too soon, restarting in %v", typ, restartInterval)
|
||||||
|
if this.flags.crashOnError {
|
||||||
|
panic("failed too soon")
|
||||||
|
}
|
||||||
timer := time.NewTimer(restartInterval)
|
timer := time.NewTimer(restartInterval)
|
||||||
select {
|
select {
|
||||||
case <- timer.C:
|
case <- timer.C:
|
||||||
@@ -329,6 +356,9 @@ func (this *environment) runRunnable(ctx context.Context, actor Runnable) (stopE
|
|||||||
err := actor.Reset(ctx)
|
err := actor.Reset(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("XXX [%s] failed to reset", typ)
|
log.Printf("XXX [%s] failed to reset", typ)
|
||||||
|
if this.flags.crashOnError {
|
||||||
|
panic("failed to reset")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
if this.Verb() { log.Printf(".// [%s] reset", typ) }
|
if this.Verb() { log.Printf(".// [%s] reset", typ) }
|
||||||
|
|||||||
35
examples/panic/main.go
Normal file
35
examples/panic/main.go
Normal file
@@ -0,0 +1,35 @@
|
|||||||
|
// Example panic demonstrates how the environment can restart actors if they
|
||||||
|
// fail.
|
||||||
|
package main
|
||||||
|
|
||||||
|
import "log"
|
||||||
|
import "time"
|
||||||
|
import "errors"
|
||||||
|
import "context"
|
||||||
|
import "math/rand"
|
||||||
|
import "git.tebibyte.media/sashakoshka/camfish"
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
camfish.Run("panic",
|
||||||
|
"Example panic demonstrates how the environment can restart " +
|
||||||
|
"actors if they fail",
|
||||||
|
new(actor))
|
||||||
|
}
|
||||||
|
|
||||||
|
// actor is an incorrectly implemented actor that panics and errs randomly.
|
||||||
|
type actor struct { }
|
||||||
|
var _ camfish.Runnable = new(actor)
|
||||||
|
func (this *actor) Type() string { return "panic" }
|
||||||
|
|
||||||
|
func (this *actor) Run(ctx context.Context) error {
|
||||||
|
log.Println("(i) [panic] panicking in 10 seconds")
|
||||||
|
select {
|
||||||
|
case <- ctx.Done(): return ctx.Err()
|
||||||
|
case <- time.After(time.Second * 10):
|
||||||
|
if rand.Int() % 2 == 0 {
|
||||||
|
panic("this is a panic")
|
||||||
|
} else {
|
||||||
|
return errors.New("this is an error")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
36
phases.go
36
phases.go
@@ -6,6 +6,7 @@ import "log"
|
|||||||
import "io/fs"
|
import "io/fs"
|
||||||
import "errors"
|
import "errors"
|
||||||
import "context"
|
import "context"
|
||||||
|
import "runtime"
|
||||||
import "strings"
|
import "strings"
|
||||||
import "path/filepath"
|
import "path/filepath"
|
||||||
import "git.tebibyte.media/sashakoshka/go-cli"
|
import "git.tebibyte.media/sashakoshka/go-cli"
|
||||||
@@ -18,12 +19,14 @@ func (this *environment) phase10FlagParsing() bool {
|
|||||||
name: this.name,
|
name: this.name,
|
||||||
description: this.description,
|
description: this.description,
|
||||||
}
|
}
|
||||||
flagHelp := set.Flag('h', "help", "Display usage information and exit", nil)
|
flagHelp := set.Flag('h', "help", "Display usage information and exit", nil)
|
||||||
flagPidFile := set.Flag('p', "pid-file", "Write the PID to the specified file", cli.ValString)
|
flagPidFile := set.Flag('p', "pid-file", "Write the PID to the specified file", cli.ValString)
|
||||||
flagUser := set.Flag('u', "user", "The user:group to run as", cli.ValString)
|
flagUser := set.Flag('u', "user", "The user:group to run as", cli.ValString)
|
||||||
flagLogDirectory := set.Flag('l', "log-directory", "Write logs to the specified directory", cli.ValString)
|
flagLogDirectory := set.Flag('l', "log-directory", "Write logs to the specified directory", cli.ValString)
|
||||||
flagConfigFile := set.Flag('c', "config-file", "Use this configuration file", cli.ValString)
|
flagConfigFile := set.Flag('c', "config-file", "Use this configuration file", cli.ValString)
|
||||||
flagVerbose := set.Flag('v', "verbose", "Enable verbose output/logging", nil)
|
flagVerbose := set.Flag('v', "verbose", "Enable verbose output/logging", nil)
|
||||||
|
flagCrash := set.Flag(0, "crash", "Crash when an actor panics", nil)
|
||||||
|
flagCrashOnError := set.Flag(0, "crash-on-error", "Crash when an actor experiences any error", nil)
|
||||||
|
|
||||||
// ask actors to add flags
|
// ask actors to add flags
|
||||||
actors, done := this.actors.RBorrow()
|
actors, done := this.actors.RBorrow()
|
||||||
@@ -60,6 +63,13 @@ func (this *environment) phase10FlagParsing() bool {
|
|||||||
if _, ok := flagVerbose.First(); ok {
|
if _, ok := flagVerbose.First(); ok {
|
||||||
this.flags.verbose = true
|
this.flags.verbose = true
|
||||||
}
|
}
|
||||||
|
if _, ok := flagCrash.First(); ok {
|
||||||
|
this.flags.crash = true
|
||||||
|
}
|
||||||
|
if _, ok := flagCrashOnError.First(); ok {
|
||||||
|
this.flags.crash = true
|
||||||
|
this.flags.crashOnError = true
|
||||||
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -128,6 +138,12 @@ func (this *environment) phase30ConfigurationParsing() bool {
|
|||||||
if this.flags.configFile != "" {
|
if this.flags.configFile != "" {
|
||||||
paths = append(paths, this.flags.configFile)
|
paths = append(paths, this.flags.configFile)
|
||||||
}
|
}
|
||||||
|
if this.Verb() {
|
||||||
|
log.Println("(i) (30) have configuration files:")
|
||||||
|
for _, paths := range paths {
|
||||||
|
log.Println("(i) (30) -", paths)
|
||||||
|
}
|
||||||
|
}
|
||||||
// parse every config and merge them all
|
// parse every config and merge them all
|
||||||
configs := make([]iniConfig, 0, len(paths))
|
configs := make([]iniConfig, 0, len(paths))
|
||||||
for _, path := range paths {
|
for _, path := range paths {
|
||||||
@@ -248,6 +264,9 @@ func (this *environment) phase70_5Trimming() bool {
|
|||||||
}()
|
}()
|
||||||
if err := this.trimActors(this.ctx, trimmable...); err != nil {
|
if err := this.trimActors(this.ctx, trimmable...); err != nil {
|
||||||
log.Println(".// (70.5) failed to trim:", err)
|
log.Println(".// (70.5) failed to trim:", err)
|
||||||
|
if this.flags.crashOnError {
|
||||||
|
panic(err)
|
||||||
|
}
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
if this.Verb() { log.Println(".// (70.5) trimmed") }
|
if this.Verb() { log.Println(".// (70.5) trimmed") }
|
||||||
@@ -263,6 +282,11 @@ func (this *environment) phase80Shutdown() bool {
|
|||||||
<- ctx.Done()
|
<- ctx.Done()
|
||||||
if errors.Is(context.Cause(ctx), context.DeadlineExceeded) {
|
if errors.Is(context.Cause(ctx), context.DeadlineExceeded) {
|
||||||
log.Println("XXX (80) shutdown timeout expired, performing emergency halt")
|
log.Println("XXX (80) shutdown timeout expired, performing emergency halt")
|
||||||
|
if Verb() || this.flags.crashOnError {
|
||||||
|
dumpBuffer := make([]byte, 8192)
|
||||||
|
runtime.Stack(dumpBuffer, true)
|
||||||
|
log.Printf("XXX (80) stack trace of all goroutines:\n%s", dumpBuffer)
|
||||||
|
}
|
||||||
log.Printf("====== [%s] END =======", this.name)
|
log.Printf("====== [%s] END =======", this.name)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
|||||||
25
util.go
25
util.go
@@ -12,6 +12,19 @@ import "strings"
|
|||||||
import "context"
|
import "context"
|
||||||
import "sync/atomic"
|
import "sync/atomic"
|
||||||
import "unicode/utf8"
|
import "unicode/utf8"
|
||||||
|
import "runtime/debug"
|
||||||
|
|
||||||
|
func panicErr(message any, stack []byte) (err error) {
|
||||||
|
if panErr, ok := message.(error); ok {
|
||||||
|
err = panErr
|
||||||
|
} else {
|
||||||
|
err = errors.New(fmt.Sprint(message))
|
||||||
|
}
|
||||||
|
if stack != nil {
|
||||||
|
err = fmt.Errorf("%w: %s", err, stack)
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
func defaul[T comparable](value, def T) T {
|
func defaul[T comparable](value, def T) T {
|
||||||
var zero T
|
var zero T
|
||||||
@@ -22,11 +35,7 @@ func defaul[T comparable](value, def T) T {
|
|||||||
func panicWrap(f func() error) (err error) {
|
func panicWrap(f func() error) (err error) {
|
||||||
defer func () {
|
defer func () {
|
||||||
if pan := recover(); pan != nil {
|
if pan := recover(); pan != nil {
|
||||||
if panErr, ok := pan.(error); ok {
|
err = panicErr(pan, debug.Stack())
|
||||||
err = panErr
|
|
||||||
} else {
|
|
||||||
err = errors.New(fmt.Sprint(pan))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} ()
|
} ()
|
||||||
|
|
||||||
@@ -37,11 +46,7 @@ func panicWrap(f func() error) (err error) {
|
|||||||
func panicWrapCtx(ctx context.Context, f func(context.Context) error) (err error) {
|
func panicWrapCtx(ctx context.Context, f func(context.Context) error) (err error) {
|
||||||
defer func () {
|
defer func () {
|
||||||
if pan := recover(); pan != nil {
|
if pan := recover(); pan != nil {
|
||||||
if panErr, ok := pan.(error); ok {
|
err = panicErr(pan, debug.Stack())
|
||||||
err = panErr
|
|
||||||
} else {
|
|
||||||
err = errors.New(fmt.Sprint(pan))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} ()
|
} ()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user