Refactoring of kubernetes informers and local caches
This commit is contained in:
Manuel de Brito Fontes
parent
8975800740
commit
e9a00ff916
23 changed files with 1704 additions and 817 deletions
|
|
@ -18,14 +18,12 @@ package controller
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
|
@ -34,9 +32,9 @@ import (
|
|||
|
||||
"github.com/golang/glog"
|
||||
|
||||
proxyproto "github.com/armon/go-proxyproto"
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
extensions "k8s.io/api/extensions/v1beta1"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/client-go/kubernetes/scheme"
|
||||
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
|
||||
"k8s.io/client-go/tools/record"
|
||||
|
|
@ -49,10 +47,9 @@ import (
|
|||
"k8s.io/ingress-nginx/internal/ingress/annotations/class"
|
||||
ngx_config "k8s.io/ingress-nginx/internal/ingress/controller/config"
|
||||
"k8s.io/ingress-nginx/internal/ingress/controller/process"
|
||||
"k8s.io/ingress-nginx/internal/ingress/controller/store"
|
||||
ngx_template "k8s.io/ingress-nginx/internal/ingress/controller/template"
|
||||
"k8s.io/ingress-nginx/internal/ingress/defaults"
|
||||
"k8s.io/ingress-nginx/internal/ingress/status"
|
||||
"k8s.io/ingress-nginx/internal/ingress/store"
|
||||
ing_net "k8s.io/ingress-nginx/internal/net"
|
||||
"k8s.io/ingress-nginx/internal/net/dns"
|
||||
"k8s.io/ingress-nginx/internal/net/ssl"
|
||||
|
|
@ -96,16 +93,12 @@ func NewNGINXController(config *Configuration, fs file.Filesystem) *NGINXControl
|
|||
}
|
||||
|
||||
n := &NGINXController{
|
||||
backendDefaults: ngx_config.NewDefault().Backend,
|
||||
binary: ngx,
|
||||
|
||||
configmap: &apiv1.ConfigMap{},
|
||||
binary: ngx,
|
||||
|
||||
isIPV6Enabled: ing_net.IsIPv6Enabled(),
|
||||
|
||||
resolver: h,
|
||||
cfg: config,
|
||||
sslCertTracker: store.NewSSLCertTracker(),
|
||||
syncRateLimiter: flowcontrol.NewTokenBucketRateLimiter(config.SyncRateLimit, 1),
|
||||
|
||||
recorder: eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{
|
||||
|
|
@ -113,6 +106,8 @@ func NewNGINXController(config *Configuration, fs file.Filesystem) *NGINXControl
|
|||
}),
|
||||
|
||||
stopCh: make(chan struct{}),
|
||||
updateCh: make(chan store.Event, 1024),
|
||||
|
||||
stopLock: &sync.Mutex{},
|
||||
|
||||
fileSystem: fs,
|
||||
|
|
@ -121,19 +116,27 @@ func NewNGINXController(config *Configuration, fs file.Filesystem) *NGINXControl
|
|||
runningConfig: &ingress.Configuration{},
|
||||
}
|
||||
|
||||
n.listers, n.controllers = n.createListers(n.stopCh)
|
||||
n.store = store.New(true,
|
||||
config.Namespace,
|
||||
config.ConfigMapName,
|
||||
config.TCPConfigMapName,
|
||||
config.UDPConfigMapName,
|
||||
config.ResyncPeriod,
|
||||
config.Client,
|
||||
fs,
|
||||
n.updateCh)
|
||||
|
||||
n.stats = newStatsCollector(config.Namespace, class.IngressClass, n.binary, n.cfg.ListenPorts.Status)
|
||||
|
||||
n.syncQueue = task.NewTaskQueue(n.syncIngress)
|
||||
|
||||
n.annotations = annotations.NewAnnotationExtractor(n)
|
||||
n.annotations = annotations.NewAnnotationExtractor(n.store)
|
||||
|
||||
if config.UpdateStatus {
|
||||
n.syncStatus = status.NewStatusSyncer(status.Config{
|
||||
Client: config.Client,
|
||||
PublishService: config.PublishService,
|
||||
IngressLister: n.listers.Ingress,
|
||||
IngressLister: n.store,
|
||||
ElectionID: config.ElectionID,
|
||||
IngressClass: class.IngressClass,
|
||||
DefaultIngressClass: class.DefaultClass,
|
||||
|
|
@ -186,9 +189,6 @@ Error loading new template : %v
|
|||
type NGINXController struct {
|
||||
cfg *Configuration
|
||||
|
||||
listers *ingress.StoreLister
|
||||
controllers *cacheController
|
||||
|
||||
annotations annotations.Extractor
|
||||
|
||||
recorder record.EventRecorder
|
||||
|
|
@ -197,10 +197,6 @@ type NGINXController struct {
|
|||
|
||||
syncStatus status.Sync
|
||||
|
||||
// local store of SSL certificates
|
||||
// (only certificates used in ingress)
|
||||
sslCertTracker *store.SSLCertTracker
|
||||
|
||||
syncRateLimiter flowcontrol.RateLimiter
|
||||
|
||||
// stopLock is used to enforce only a single call to Stop is active.
|
||||
|
|
@ -208,7 +204,8 @@ type NGINXController struct {
|
|||
// allowing concurrent stoppers leads to stack traces.
|
||||
stopLock *sync.Mutex
|
||||
|
||||
stopCh chan struct{}
|
||||
stopCh chan struct{}
|
||||
updateCh chan store.Event
|
||||
|
||||
// ngxErrCh channel used to detect errors with the nginx processes
|
||||
ngxErrCh chan error
|
||||
|
|
@ -220,8 +217,6 @@ type NGINXController struct {
|
|||
|
||||
t *ngx_template.Template
|
||||
|
||||
configmap *apiv1.ConfigMap
|
||||
|
||||
binary string
|
||||
resolver []net.IP
|
||||
|
||||
|
|
@ -231,14 +226,11 @@ type NGINXController struct {
|
|||
// returns true if IPV6 is enabled in the pod
|
||||
isIPV6Enabled bool
|
||||
|
||||
// returns true if proxy protocol es enabled
|
||||
IsProxyProtocolEnabled bool
|
||||
|
||||
isShuttingDown bool
|
||||
|
||||
Proxy *TCPProxy
|
||||
|
||||
backendDefaults defaults.Backend
|
||||
store store.Storer
|
||||
|
||||
fileSystem filesystem.Filesystem
|
||||
}
|
||||
|
|
@ -247,32 +239,12 @@ type NGINXController struct {
|
|||
func (n *NGINXController) Start() {
|
||||
glog.Infof("starting Ingress controller")
|
||||
|
||||
n.controllers.Run(n.stopCh)
|
||||
|
||||
// initial sync of secrets to avoid unnecessary reloads
|
||||
glog.Info("running initial sync of secrets")
|
||||
for _, obj := range n.listers.Ingress.List() {
|
||||
ing := obj.(*extensions.Ingress)
|
||||
|
||||
if !class.IsValid(ing) {
|
||||
a := ing.GetAnnotations()[class.IngressKey]
|
||||
glog.Infof("ignoring add for ingress %v based on annotation %v with value %v", ing.Name, class.IngressKey, a)
|
||||
continue
|
||||
}
|
||||
|
||||
n.readSecrets(ing)
|
||||
}
|
||||
|
||||
if n.cfg.EnableSSLChainCompletion {
|
||||
go wait.Until(n.checkSSLChainIssues, 60*time.Second, n.stopCh)
|
||||
}
|
||||
n.store.Run(n.stopCh)
|
||||
|
||||
if n.syncStatus != nil {
|
||||
go n.syncStatus.Run()
|
||||
}
|
||||
|
||||
go wait.Until(n.checkMissingSecrets, 30*time.Second, n.stopCh)
|
||||
|
||||
done := make(chan error, 1)
|
||||
cmd := exec.Command(n.binary, "-c", cfgPath)
|
||||
|
||||
|
|
@ -283,6 +255,10 @@ func (n *NGINXController) Start() {
|
|||
Pgid: 0,
|
||||
}
|
||||
|
||||
if n.cfg.EnableSSLPassthrough {
|
||||
n.setupSSLProxy()
|
||||
}
|
||||
|
||||
glog.Info("starting NGINX process...")
|
||||
n.start(cmd)
|
||||
|
||||
|
|
@ -310,6 +286,12 @@ func (n *NGINXController) Start() {
|
|||
// start a new nginx master process if the controller is not being stopped
|
||||
n.start(cmd)
|
||||
}
|
||||
case evt := <-n.updateCh:
|
||||
if n.isShuttingDown {
|
||||
break
|
||||
}
|
||||
glog.V(3).Infof("Event %v received - object %v", evt.Type, evt.Obj)
|
||||
n.syncQueue.Enqueue(evt.Obj)
|
||||
case <-n.stopCh:
|
||||
break
|
||||
}
|
||||
|
|
@ -412,37 +394,6 @@ Error: %v
|
|||
return nil
|
||||
}
|
||||
|
||||
// SetConfig sets the configured configmap
|
||||
func (n *NGINXController) SetConfig(cmap *apiv1.ConfigMap) {
|
||||
n.configmap = cmap
|
||||
n.IsProxyProtocolEnabled = false
|
||||
|
||||
m := map[string]string{}
|
||||
if cmap != nil {
|
||||
m = cmap.Data
|
||||
}
|
||||
|
||||
val, ok := m["use-proxy-protocol"]
|
||||
if ok {
|
||||
b, err := strconv.ParseBool(val)
|
||||
if err == nil {
|
||||
n.IsProxyProtocolEnabled = b
|
||||
}
|
||||
}
|
||||
|
||||
c := ngx_template.ReadConfig(m)
|
||||
if c.SSLSessionTicketKey != "" {
|
||||
d, err := base64.StdEncoding.DecodeString(c.SSLSessionTicketKey)
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error decoding key ssl-session-ticket-key: %v", err)
|
||||
c.SSLSessionTicketKey = ""
|
||||
}
|
||||
ioutil.WriteFile("/etc/nginx/tickets.key", d, 0644)
|
||||
}
|
||||
|
||||
n.backendDefaults = c.Backend
|
||||
}
|
||||
|
||||
// OnUpdate is called periodically by syncQueue to keep the configuration in sync.
|
||||
//
|
||||
// 1. converts configmap configuration to custom configuration object
|
||||
|
|
@ -452,7 +403,7 @@ func (n *NGINXController) SetConfig(cmap *apiv1.ConfigMap) {
|
|||
// returning nill implies the backend will be reloaded.
|
||||
// if an error is returned means requeue the update
|
||||
func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {
|
||||
cfg := ngx_template.ReadConfig(n.configmap.Data)
|
||||
cfg := n.store.GetBackendConfiguration()
|
||||
cfg.Resolver = n.resolver
|
||||
|
||||
servers := []*TCPServer{}
|
||||
|
|
@ -488,10 +439,6 @@ func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {
|
|||
})
|
||||
}
|
||||
|
||||
if n.cfg.EnableSSLPassthrough {
|
||||
n.Proxy.ServerList = servers
|
||||
}
|
||||
|
||||
// we need to check if the status module configuration changed
|
||||
if cfg.EnableVtsStatus {
|
||||
n.setupMonitor(vtsStatusModule)
|
||||
|
|
@ -562,63 +509,48 @@ func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {
|
|||
|
||||
setHeaders := map[string]string{}
|
||||
if cfg.ProxySetHeaders != "" {
|
||||
cmap, exists, err := n.listers.ConfigMap.GetByKey(cfg.ProxySetHeaders)
|
||||
cmap, err := n.store.GetConfigMap(cfg.ProxySetHeaders)
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error reading configmap %v: %v", cfg.ProxySetHeaders, err)
|
||||
}
|
||||
|
||||
if exists {
|
||||
setHeaders = cmap.(*apiv1.ConfigMap).Data
|
||||
}
|
||||
setHeaders = cmap.Data
|
||||
}
|
||||
|
||||
addHeaders := map[string]string{}
|
||||
if cfg.AddHeaders != "" {
|
||||
cmap, exists, err := n.listers.ConfigMap.GetByKey(cfg.AddHeaders)
|
||||
cmap, err := n.store.GetConfigMap(cfg.AddHeaders)
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error reading configmap %v: %v", cfg.AddHeaders, err)
|
||||
}
|
||||
|
||||
if exists {
|
||||
addHeaders = cmap.(*apiv1.ConfigMap).Data
|
||||
}
|
||||
addHeaders = cmap.Data
|
||||
}
|
||||
|
||||
sslDHParam := ""
|
||||
if cfg.SSLDHParam != "" {
|
||||
secretName := cfg.SSLDHParam
|
||||
s, exists, err := n.listers.Secret.GetByKey(secretName)
|
||||
|
||||
secret, err := n.store.GetSecret(secretName)
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error reading secret %v: %v", secretName, err)
|
||||
}
|
||||
|
||||
if exists {
|
||||
secret := s.(*apiv1.Secret)
|
||||
nsSecName := strings.Replace(secretName, "/", "-", -1)
|
||||
nsSecName := strings.Replace(secretName, "/", "-", -1)
|
||||
|
||||
dh, ok := secret.Data["dhparam.pem"]
|
||||
if ok {
|
||||
pemFileName, err := ssl.AddOrUpdateDHParam(nsSecName, dh)
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error adding or updating dhparam %v file: %v", nsSecName, err)
|
||||
} else {
|
||||
sslDHParam = pemFileName
|
||||
}
|
||||
dh, ok := secret.Data["dhparam.pem"]
|
||||
if ok {
|
||||
pemFileName, err := ssl.AddOrUpdateDHParam(nsSecName, dh, n.fileSystem)
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error adding or updating dhparam %v file: %v", nsSecName, err)
|
||||
} else {
|
||||
sslDHParam = pemFileName
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cfg.SSLDHParam = sslDHParam
|
||||
|
||||
// disable features are not available in some platforms
|
||||
switch runtime.GOARCH {
|
||||
case "arm", "arm64", "ppc64le":
|
||||
cfg.EnableModsecurity = false
|
||||
case "s390x":
|
||||
cfg.EnableModsecurity = false
|
||||
cfg.EnableBrotli = false
|
||||
}
|
||||
|
||||
tc := ngx_config.TemplateConfig{
|
||||
ProxySetHeaders: setHeaders,
|
||||
AddHeaders: addHeaders,
|
||||
|
|
@ -713,3 +645,49 @@ func nextPowerOf2(v int) int {
|
|||
|
||||
return v
|
||||
}
|
||||
|
||||
func (n *NGINXController) setupSSLProxy() {
|
||||
sslPort := n.cfg.ListenPorts.HTTPS
|
||||
proxyPort := n.cfg.ListenPorts.SSLProxy
|
||||
|
||||
glog.Info("starting TLS proxy for SSL passthrough")
|
||||
n.Proxy = &TCPProxy{
|
||||
Default: &TCPServer{
|
||||
Hostname: "localhost",
|
||||
IP: "127.0.0.1",
|
||||
Port: proxyPort,
|
||||
ProxyProtocol: true,
|
||||
},
|
||||
}
|
||||
|
||||
listener, err := net.Listen("tcp", fmt.Sprintf(":%v", sslPort))
|
||||
if err != nil {
|
||||
glog.Fatalf("%v", err)
|
||||
}
|
||||
|
||||
proxyList := &proxyproto.Listener{Listener: listener}
|
||||
|
||||
// start goroutine that accepts tcp connections in port 443
|
||||
go func() {
|
||||
for {
|
||||
var conn net.Conn
|
||||
var err error
|
||||
|
||||
if n.store.GetBackendConfiguration().UseProxyProtocol {
|
||||
// we need to wrap the listener in order to decode
|
||||
// proxy protocol before handling the connection
|
||||
conn, err = proxyList.Accept()
|
||||
} else {
|
||||
conn, err = listener.Accept()
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
glog.Warningf("unexpected error accepting tcp connection: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
glog.V(3).Infof("remote address %s to local %s", conn.RemoteAddr(), conn.LocalAddr())
|
||||
go n.Proxy.Handle(conn)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue