-
Notifications
You must be signed in to change notification settings - Fork 356
/
main.go
159 lines (132 loc) · 3.69 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
package main
import (
"context"
"fmt"
"net/http"
_ "net/http/pprof"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
_ "k8s.io/client-go/plugin/pkg/client/auth/oidc"
"github.com/spotahome/redis-operator/cmd/utils"
"github.com/spotahome/redis-operator/log"
"github.com/spotahome/redis-operator/metrics"
"github.com/spotahome/redis-operator/operator/redisfailover"
"github.com/spotahome/redis-operator/service/k8s"
"github.com/spotahome/redis-operator/service/redis"
)
const (
gracePeriod = 5 * time.Second
metricsNamespace = "redis_operator"
)
// Main is the main runner.
type Main struct {
flags *utils.CMDFlags
logger log.Logger
stopC chan struct{}
}
// New returns a Main object.
func New(logger log.Logger) Main {
// Init flags.
flgs := &utils.CMDFlags{}
flgs.Init()
return Main{
logger: logger,
flags: flgs,
}
}
// Run execs the program.
func (m *Main) Run() error {
// Create signal channels.
m.stopC = make(chan struct{})
errC := make(chan error)
// Set correct logging.
err := m.logger.Set(log.Level(strings.ToLower(m.flags.LogLevel)))
if err != nil {
return err
}
// Create the metrics client.
metricsRecorder := metrics.NewRecorder(metricsNamespace, prometheus.DefaultRegisterer)
// Serve metrics.
go func() {
log.Infof("Listening on %s for metrics exposure on URL %s", m.flags.ListenAddr, m.flags.MetricsPath)
http.Handle(m.flags.MetricsPath, promhttp.Handler())
err := http.ListenAndServe(m.flags.ListenAddr, nil)
if err != nil {
log.Fatal(err)
}
}()
// Kubernetes clients.
k8sClient, customClient, aeClientset, err := utils.CreateKubernetesClients(m.flags)
if err != nil {
return err
}
// Create kubernetes service.
k8sservice := k8s.New(k8sClient, customClient, aeClientset, m.logger, metricsRecorder)
// Create the redis clients
redisClient := redis.New(metricsRecorder)
// Get lease lock resource namespace
lockNamespace := getNamespace()
// Create operator and run.
redisfailoverOperator, err := redisfailover.New(m.flags.ToRedisOperatorConfig(), k8sservice, k8sClient, lockNamespace, redisClient, metricsRecorder, m.logger)
if err != nil {
return err
}
go func() {
errC <- redisfailoverOperator.Run(context.Background())
}()
// Await signals.
sigC := m.createSignalCapturer()
var finalErr error
select {
case <-sigC:
m.logger.Infof("Signal captured, exiting...")
case err := <-errC:
m.logger.Errorf("Error received: %s, exiting...", err)
finalErr = err
}
m.stop(m.stopC)
return finalErr
}
func (m *Main) createSignalCapturer() <-chan os.Signal {
sigC := make(chan os.Signal, 1)
signal.Notify(sigC, syscall.SIGTERM, syscall.SIGINT)
return sigC
}
func (m *Main) stop(stopC chan struct{}) {
m.logger.Infof("Stopping everything, waiting %s...", gracePeriod)
// stop everything and let them time to stop
close(stopC)
time.Sleep(gracePeriod)
}
func getNamespace() string {
// This way assumes you've set the POD_NAMESPACE environment
// variable using the downward API. This check has to be done first
// for backwards compatibility with the way InClusterConfig was
// originally set up
if ns, ok := os.LookupEnv("POD_NAMESPACE"); ok {
return ns
}
// Fall back to the namespace associated with the service account
// token, if available
if data, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil {
if ns := strings.TrimSpace(string(data)); len(ns) > 0 {
return ns
}
}
return "default"
}
// Run app.
func main() {
logger := log.Base()
m := New(logger)
if err := m.Run(); err != nil {
fmt.Fprintf(os.Stderr, "error executing: %s", err)
os.Exit(1)
}
os.Exit(0)
}