-
Notifications
You must be signed in to change notification settings - Fork 241
/
main.go
144 lines (127 loc) · 4.52 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
package main
import (
"context"
"errors"
"flag"
"fmt"
"os"
"os/exec"
"os/signal"
"syscall"
"time"
"github.com/fluent/fluent-operator/v2/pkg/filenotify"
"github.com/fsnotify/fsnotify"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"golang.org/x/sync/errgroup"
)
const (
defaultBinPath = "/fluent-bit/bin/fluent-bit"
defaultCfgPath = "/fluent-bit/etc/fluent-bit.conf"
defaultWatchDir = "/fluent-bit/config"
defaultPollInterval = 1 * time.Second
)
func main() {
var configPath string
var externalPluginPath string
var binPath string
var watchPath string
var poll bool
var pollInterval time.Duration
flag.StringVar(&binPath, "b", defaultBinPath, "The fluent bit binary path.")
flag.StringVar(&configPath, "c", defaultCfgPath, "The config file path.")
flag.StringVar(&externalPluginPath, "e", "", "Path to external plugin (shared lib)")
flag.StringVar(&watchPath, "watch-path", defaultWatchDir, "The path to watch.")
flag.BoolVar(&poll, "poll", false, "Use poll watcher instead of ionotify.")
flag.DurationVar(&pollInterval, "poll-interval", defaultPollInterval, "Poll interval if using poll watcher.")
// Deprecated flags to be removed in one of the next releases.
var exitOnFailure bool
var flbTerminationTimeout time.Duration
flag.BoolVar(&exitOnFailure, "exit-on-failure", false, "Deprecated: This has no effect anymore.")
flag.DurationVar(&flbTerminationTimeout, "flb-timeout", 0, "Deprecated: This has no effect anymore.")
flag.Parse()
signalCtx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer cancel()
logger := log.NewLogfmtLogger(os.Stdout)
logger = log.With(logger, "time", log.TimestampFormat(time.Now, time.RFC3339))
if exitOnFailure {
level.Warn(logger).Log("--exit-on-failure is deprecated. The process will exit no matter what if fluent-bit exits so this can safely be removed.")
}
if flbTerminationTimeout > 0 {
level.Warn(logger).Log("--flb-timeout is deprecated. Consider setting the terminationGracePeriod field on the `(Cluster)FluentBit` instance.")
}
// First, launch the fluent-bit process.
args := []string{"--enable-hot-reload", "-c", configPath}
if externalPluginPath != "" {
args = append(args, "-e", externalPluginPath)
}
cmd := exec.Command(binPath, args...)
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
if err := cmd.Start(); err != nil {
_ = level.Error(logger).Log("msg", "failed to start fluent-bit", "error", err)
os.Exit(1)
}
_ = level.Info(logger).Log("msg", "fluent-bit started")
grp, grpCtx := errgroup.WithContext(context.Background())
grp.Go(func() error {
// Watch the process. If it exits, we want to crash immediately.
defer cancel()
if err := cmd.Wait(); err != nil {
return fmt.Errorf("failed to run fluent-bit: %w", err)
}
return nil
})
grp.Go(func() error {
// Watch the config as it's loaded into the pod and trigger a config reload.
var watcher filenotify.FileWatcher
if poll {
watcher = filenotify.NewPollingWatcher(pollInterval)
} else {
var err error
watcher, err = filenotify.NewEventWatcher()
if err != nil {
return fmt.Errorf("failed to open event watcher: %w", err)
}
}
if err := watcher.Add(watchPath); err != nil {
return fmt.Errorf("failed to watch path %q: %w", watchPath, err)
}
for {
select {
case <-signalCtx.Done():
return nil
case <-grpCtx.Done():
return nil
case event := <-watcher.Events():
if !isValidEvent(event) {
continue
}
_ = level.Info(logger).Log("msg", "Config file changed, reloading...")
if err := cmd.Process.Signal(syscall.SIGHUP); err != nil {
return fmt.Errorf("failed to reload config: %w", err)
}
case err := <-watcher.Errors():
return fmt.Errorf("failed the watcher: %w", err)
}
}
})
select {
case <-signalCtx.Done():
case <-grpCtx.Done():
}
// Always try to gracefully shut down fluent-bit. This will allow `cmd.Wait` above to finish
// and thus allow `grp.Wait` below to return.
if err := cmd.Process.Signal(syscall.SIGTERM); err != nil && !errors.Is(err, os.ErrProcessDone) {
_ = level.Error(logger).Log("msg", "Failed to send SIGTERM to fluent-bit", "error", err)
// Do not exit on error here. The process might've died and that's okay.
}
if err := grp.Wait(); err != nil {
_ = level.Error(logger).Log("msg", "Failure during the run time of fluent-bit", "error", err)
os.Exit(1)
}
}
// Inspired by https://github.com/jimmidyson/configmap-reload
func isValidEvent(event fsnotify.Event) bool {
return event.Op == fsnotify.Create || event.Op == fsnotify.Write
}