Skip to content

Commit 8ff7632

Browse files
enocommarkmandel
enocom
authored and
markmandel
committed
Simplify cmd/controller/main.go
- Extract ctlConfig type to store configuration - Move viper and pflag calls into separate function - Ensure all controllers have same method signature for `Run` - Use runner interface to start all controllers - Create healthServer to match controllers - Use Run signature to match Kubernetes precedent
1 parent b8d38ca commit 8ff7632

7 files changed

Lines changed: 126 additions & 99 deletions

File tree

‎cmd/controller/main.go‎

Lines changed: 117 additions & 90 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919
"net/http"
2020
"os"
2121
"path/filepath"
22+
"reflect"
2223
"strings"
2324
"time"
2425

@@ -43,21 +44,82 @@ import (
4344
)
4445

4546
const (
46-
sidecarFlag = "sidecar"
47-
pullSidecarFlag = "always-pull-sidecar"
48-
minPortFlag = "min-port"
49-
maxPortFlag = "max-port"
50-
certFileFlag = "cert-file"
51-
keyFileFlag = "key-file"
52-
controllerThreadiness = 2
47+
sidecarFlag = "sidecar"
48+
pullSidecarFlag = "always-pull-sidecar"
49+
minPortFlag = "min-port"
50+
maxPortFlag = "max-port"
51+
certFileFlag = "cert-file"
52+
keyFileFlag = "key-file"
53+
workers = 2
54+
defaultResync = 30 * time.Second
5355
)
5456

5557
var (
5658
logger = runtime.NewLoggerWithSource("main")
5759
)
5860

5961
// main starts the operator for the gameserver CRD
60-
func main() { // nolint: gocyclo
62+
func main() {
63+
ctlConf := parseEnvFlags()
64+
if err := ctlConf.validate(); err != nil {
65+
logger.WithError(err).Fatal("Could not create controller from environment or flags")
66+
}
67+
68+
clientConf, err := rest.InClusterConfig()
69+
if err != nil {
70+
logger.WithError(err).Fatal("Could not create in cluster config")
71+
}
72+
73+
kubeClient, err := kubernetes.NewForConfig(clientConf)
74+
if err != nil {
75+
logger.WithError(err).Fatal("Could not create the kubernetes clientset")
76+
}
77+
78+
extClient, err := extclientset.NewForConfig(clientConf)
79+
if err != nil {
80+
logger.WithError(err).Fatal("Could not create the api extension clientset")
81+
}
82+
83+
agonesClient, err := versioned.NewForConfig(clientConf)
84+
if err != nil {
85+
logger.WithError(err).Fatal("Could not create the agones api clientset")
86+
}
87+
88+
health := healthcheck.NewHandler()
89+
wh := webhooks.NewWebHook(ctlConf.certFile, ctlConf.keyFile)
90+
agonesInformerFactory := externalversions.NewSharedInformerFactory(agonesClient, defaultResync)
91+
kubeInformationFactory := informers.NewSharedInformerFactory(kubeClient, defaultResync)
92+
93+
gsController := gameservers.NewController(wh, health,
94+
ctlConf.minPort, ctlConf.maxPort, ctlConf.sidecarImage, ctlConf.alwaysPullSidecar,
95+
kubeClient, kubeInformationFactory, extClient, agonesClient, agonesInformerFactory)
96+
gsSetController := gameserversets.NewController(wh, health,
97+
kubeClient, extClient, agonesClient, agonesInformerFactory)
98+
fleetController := fleets.NewController(wh, health,
99+
kubeClient, extClient, agonesClient, agonesInformerFactory)
100+
faController := fleetallocation.NewController(wh, kubeClient, extClient, agonesClient, agonesInformerFactory)
101+
102+
stop := signals.NewStopChannel()
103+
104+
kubeInformationFactory.Start(stop)
105+
agonesInformerFactory.Start(stop)
106+
107+
rs := []runner{
108+
wh, gsController, gsSetController, fleetController, faController, healthServer{handler: health},
109+
}
110+
for _, r := range rs {
111+
go func(rr runner) {
112+
if runErr := rr.Run(workers, stop); runErr != nil {
113+
logger.WithError(runErr).Fatalf("could not start runner: %s", reflect.TypeOf(rr))
114+
}
115+
}(r)
116+
}
117+
118+
<-stop
119+
logger.Info("Shut down agones controllers")
120+
}
121+
122+
func parseEnvFlags() config {
61123
exec, err := os.Executable()
62124
if err != nil {
63125
logger.WithError(err).Fatal("Could not get executable path")
@@ -101,95 +163,60 @@ func main() { // nolint: gocyclo
101163
WithField("alwaysPullSidecarImage", alwaysPullSidecar).
102164
WithField("Version", pkg.Version).Info("starting gameServer operator...")
103165

104-
if minPort <= 0 || maxPort <= 0 {
105-
logger.Fatal("Min Port and Max Port values are required.")
106-
} else if maxPort < minPort {
107-
logger.Fatal("Max Port cannot be set less that the Min Port")
108-
}
109-
110-
config, err := rest.InClusterConfig()
111-
if err != nil {
112-
logger.WithError(err).Fatal("Could not create in cluster config")
166+
return config{
167+
minPort: minPort,
168+
maxPort: maxPort,
169+
sidecarImage: sidecarImage,
170+
alwaysPullSidecar: alwaysPullSidecar,
171+
keyFile: keyFile,
172+
certFile: certFile,
113173
}
174+
}
114175

115-
kubeClient, err := kubernetes.NewForConfig(config)
116-
if err != nil {
117-
logger.WithError(err).Fatal("Could not create the kubernetes clientset")
118-
}
176+
// config stores all required configuration to create a game server controller.
177+
type config struct {
178+
minPort int32
179+
maxPort int32
180+
sidecarImage string
181+
alwaysPullSidecar bool
182+
keyFile string
183+
certFile string
184+
}
119185

120-
extClient, err := extclientset.NewForConfig(config)
121-
if err != nil {
122-
logger.WithError(err).Fatal("Could not create the api extension clientset")
186+
// validate ensures the ctlConfig data is valid.
187+
func (c config) validate() error {
188+
if c.minPort <= 0 || c.maxPort <= 0 {
189+
return errors.New("min Port and Max Port values are required")
123190
}
124-
125-
agonesClient, err := versioned.NewForConfig(config)
126-
if err != nil {
127-
logger.WithError(err).Fatal("Could not create the agones api clientset")
191+
if c.maxPort < c.minPort {
192+
return errors.New("max Port cannot be set less that the Min Port")
128193
}
194+
return nil
195+
}
129196

130-
health := healthcheck.NewHandler()
131-
wh := webhooks.NewWebHook(certFile, keyFile)
132-
agonesInformerFactory := externalversions.NewSharedInformerFactory(agonesClient, 30*time.Second)
133-
kubeInformationFactory := informers.NewSharedInformerFactory(kubeClient, 30*time.Second)
134-
135-
gsController := gameservers.NewController(wh, health, minPort, maxPort, sidecarImage, alwaysPullSidecar, kubeClient, kubeInformationFactory, extClient, agonesClient, agonesInformerFactory)
136-
gsSetController := gameserversets.NewController(wh, health, kubeClient, extClient, agonesClient, agonesInformerFactory)
137-
fleetController := fleets.NewController(wh, health, kubeClient, extClient, agonesClient, agonesInformerFactory)
138-
faController := fleetallocation.NewController(wh, kubeClient, extClient, agonesClient, agonesInformerFactory)
139-
140-
stop := signals.NewStopChannel()
141-
142-
kubeInformationFactory.Start(stop)
143-
agonesInformerFactory.Start(stop)
197+
type runner interface {
198+
Run(workers int, stop <-chan struct{}) error
199+
}
144200

145-
go func() {
146-
if err := wh.Run(stop); err != nil { // nolint: vetshadow
147-
logger.WithError(err).Fatal("could not run webhook server")
148-
}
149-
}()
150-
go func() {
151-
err = gsController.Run(controllerThreadiness, stop)
152-
if err != nil {
153-
logger.WithError(err).Fatal("Could not run gameserver controller")
154-
}
155-
}()
156-
go func() {
157-
err = gsSetController.Run(controllerThreadiness, stop)
158-
if err != nil {
159-
logger.WithError(err).Fatal("Could not run gameserverset controller")
160-
}
161-
}()
162-
go func() {
163-
err = fleetController.Run(controllerThreadiness, stop)
164-
if err != nil {
165-
logger.WithError(err).Fatal("Could not run fleet controller")
166-
}
167-
}()
168-
go func() {
169-
err = faController.Run(stop)
170-
if err != nil {
171-
logger.WithError(err).Fatal("Could not run fleet controller")
172-
}
173-
}()
201+
type healthServer struct {
202+
handler http.Handler
203+
}
174204

175-
go func() {
176-
logger.Info("Starting health check...")
177-
srv := &http.Server{
178-
Addr: ":8080",
179-
Handler: health,
180-
}
181-
defer srv.Close() // nolint: errcheck
182-
183-
if err := srv.ListenAndServe(); err != nil {
184-
if err == http.ErrServerClosed {
185-
logger.WithError(err).Info("health check: http server closed")
186-
} else {
187-
err := errors.Wrap(err, "Could not listen on :8080")
188-
runtime.HandleError(logger.WithError(err), err)
189-
}
205+
func (h healthServer) Run(workers int, stop <-chan struct{}) error {
206+
logger.Info("Starting health check...")
207+
srv := &http.Server{
208+
Addr: ":8080",
209+
Handler: h.handler,
210+
}
211+
defer srv.Close() // nolint: errcheck
212+
213+
if err := srv.ListenAndServe(); err != nil {
214+
if err == http.ErrServerClosed {
215+
logger.WithError(err).Info("health check: http server closed")
216+
} else {
217+
wrappedErr := errors.Wrap(err, "Could not listen on :8080")
218+
runtime.HandleError(logger.WithError(wrappedErr), wrappedErr)
190219
}
191-
}()
192-
193-
<-stop
194-
logger.Info("Shut down agones controllers")
220+
}
221+
return nil
195222
}

‎pkg/fleetallocation/controller.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ func NewController(
104104

105105
// Run runs this controller. This controller doesn't (currently)
106106
// have a worker/queue, and as such, does not block.
107-
func (c *Controller) Run(stop <-chan struct{}) error {
107+
func (c *Controller) Run(workers int, stop <-chan struct{}) error {
108108
err := crd.WaitForEstablishedCRD(c.crdGetter, "fleetallocations."+stable.GroupName, c.logger)
109109
if err != nil {
110110
return err

‎pkg/fleets/controller.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ func (c *Controller) creationMutationHandler(review admv1beta1.AdmissionReview)
160160

161161
// Run the Fleet controller. Will block until stop is closed.
162162
// Runs threadiness number workers to process the rate limited queue
163-
func (c *Controller) Run(threadiness int, stop <-chan struct{}) error {
163+
func (c *Controller) Run(workers int, stop <-chan struct{}) error {
164164
err := crd.WaitForEstablishedCRD(c.crdGetter, "fleets.stable.agones.dev", c.logger)
165165
if err != nil {
166166
return err
@@ -171,7 +171,7 @@ func (c *Controller) Run(threadiness int, stop <-chan struct{}) error {
171171
return errors.New("failed to wait for caches to sync")
172172
}
173173

174-
c.workerqueue.Run(threadiness, stop)
174+
c.workerqueue.Run(workers, stop)
175175
return nil
176176
}
177177

‎pkg/gameservers/controller.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -220,7 +220,7 @@ func (c *Controller) creationValidationHandler(review admv1beta1.AdmissionReview
220220

221221
// Run the GameServer controller. Will block until stop is closed.
222222
// Runs threadiness number workers to process the rate limited queue
223-
func (c *Controller) Run(threadiness int, stop <-chan struct{}) error {
223+
func (c *Controller) Run(workers int, stop <-chan struct{}) error {
224224
err := crd.WaitForEstablishedCRD(c.crdGetter, "gameservers.stable.agones.dev", c.logger)
225225
if err != nil {
226226
return err
@@ -239,7 +239,7 @@ func (c *Controller) Run(threadiness int, stop <-chan struct{}) error {
239239
// Run the Health Controller
240240
go c.healthController.Run(stop)
241241

242-
c.workerqueue.Run(threadiness, stop)
242+
c.workerqueue.Run(workers, stop)
243243
return nil
244244
}
245245

‎pkg/gameserversets/controller.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ func NewController(
126126

127127
// Run the GameServerSet controller. Will block until stop is closed.
128128
// Runs threadiness number workers to process the rate limited queue
129-
func (c *Controller) Run(threadiness int, stop <-chan struct{}) error {
129+
func (c *Controller) Run(workers int, stop <-chan struct{}) error {
130130
err := crd.WaitForEstablishedCRD(c.crdGetter, "gameserversets."+stable.GroupName, c.logger)
131131
if err != nil {
132132
return err
@@ -137,7 +137,7 @@ func (c *Controller) Run(threadiness int, stop <-chan struct{}) error {
137137
return errors.New("failed to wait for caches to sync")
138138
}
139139

140-
c.workerqueue.Run(threadiness, stop)
140+
c.workerqueue.Run(workers, stop)
141141
return nil
142142
}
143143

‎pkg/util/webhooks/webhooks.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ func NewWebHook(certFile, keyFile string) *WebHook {
7575

7676
// Run runs the webhook server, starting a https listener.
7777
// Will block on stop channel
78-
func (wh *WebHook) Run(stop <-chan struct{}) error {
78+
func (wh *WebHook) Run(workers int, stop <-chan struct{}) error {
7979
go func() {
8080
<-stop
8181
wh.server.Close() // nolint: errcheck

‎pkg/util/webhooks/webhooks_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@ func TestWebHookAddHandler(t *testing.T) {
116116
})
117117
}
118118

119-
err := wh.Run(stop)
119+
err := wh.Run(0, stop)
120120
assert.Nil(t, err)
121121

122122
client := ts.server.Client()

0 commit comments

Comments
 (0)