DevFW-CICD/ingress-nginx-helm
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate to structured logging (klog)

Browse source
This commit is contained in:
Manuel Alejandro de Brito Fontes 2020-09-27 17:32:40 -03:00
parent 93ac8d5a21
commit 108637bb1c
31 changed files with 161 additions and 171 deletions
Download patch file Download diff file Expand all files Collapse all files

9
internal/admission/controller/main.go
View file

@ -80,8 +80,7 @@ func (ia *IngressAdmission) HandleAdmission(ar *admissionv1.AdmissionReview) {
ingress := networking.Ingress{}
deserializer := codecs.UniversalDeserializer()
if _, _, err := deserializer.Decode(ar.Request.Object.Raw, nil, &ingress); err != nil {
klog.Errorf("failed to decode ingress %s in namespace %s: %s, refusing it",
ar.Request.Name, ar.Request.Namespace, err.Error())
klog.ErrorS(err, "failed to decode ingress", "ingress", ar.Request.Name, "namespace", ar.Request.Namespace)
ar.Response = &admissionv1.AdmissionResponse{
UID: ar.Request.UID,
@ -97,8 +96,7 @@ func (ia *IngressAdmission) HandleAdmission(ar *admissionv1.AdmissionReview) {
}
if err := ia.Checker.CheckIngress(&ingress); err != nil {
klog.Errorf("failed to generate configuration for ingress %s in namespace %s: %s, refusing it",
ar.Request.Name, ar.Request.Namespace, err.Error())
klog.ErrorS(err, "failed to generate configuration for ingress", "ingress", ar.Request.Name, "namespace", ar.Request.Namespace)
ar.Response = &admissionv1.AdmissionResponse{
UID: ar.Request.UID,
Allowed: false,
@ -111,8 +109,7 @@ func (ia *IngressAdmission) HandleAdmission(ar *admissionv1.AdmissionReview) {
return
}
klog.Infof("successfully validated configuration, accepting ingress %s in namespace %s",
ar.Request.Name, ar.Request.Namespace)
klog.InfoS("successfully validated configuration, accepting", "ingress", ar.Request.Name, "namespace", ar.Request.Namespace)
ar.Response = &admissionv1.AdmissionResponse{
UID: ar.Request.UID,
Allowed: true,

4
internal/admission/controller/server.go
View file

@ -60,14 +60,14 @@ func NewAdmissionControllerServer(ac AdmissionController) *AdmissionControllerSe
func (acs *AdmissionControllerServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
review, err := parseAdmissionReview(acs.Decoder, r.Body)
if err != nil {
klog.Errorf("Unexpected error decoding request: %v", err)
klog.ErrorS(err, "Unexpected error decoding request")
w.WriteHeader(http.StatusBadRequest)
return
}
acs.AdmissionController.HandleAdmission(review)
if err := writeAdmissionReview(w, review); err != nil {
klog.Errorf("Unexpected returning admission review: %v", err)
klog.ErrorS(err, "Unexpected returning admission review")
}
}

2
internal/file/file.go
View file

@ -29,7 +29,7 @@ func SHA1(filename string) string {
hasher := sha1.New()
s, err := ioutil.ReadFile(filename)
if err != nil {
klog.Errorf("Error reading file %v", err)
klog.ErrorS(err, "Error reading file", "path", filename)
return ""
}

8
internal/ingress/annotations/annotations.go
View file

@ -175,7 +175,7 @@ func (e Extractor) Extract(ing *networking.Ingress) *Ingress {
data := make(map[string]interface{})
for name, annotationParser := range e.annotations {
val, err := annotationParser.Parse(ing)
klog.V(5).Infof("annotation %v in Ingress %v/%v: %v", name, ing.GetNamespace(), ing.GetName(), val)
klog.V(5).InfoS("Parsing Ingress annotation", "name", name, "namespace", ing.GetNamespace(), "ingress", ing.GetName(), "value", val)
if err != nil {
if errors.IsMissingAnnotations(err) {
continue
@ -197,11 +197,11 @@ func (e Extractor) Extract(ing *networking.Ingress) *Ingress {
if !alreadyDenied {
errString := err.Error()
data[DeniedKeyName] = &errString
klog.Errorf("error reading %v annotation in Ingress %v/%v: %v", name, ing.GetNamespace(), ing.GetName(), err)
klog.ErrorS(err, "error reading Ingress annotation", "name", name, "namespace", ing.GetNamespace(), "ingress", ing.GetName())
continue
}
klog.V(5).Infof("error reading %v annotation in Ingress %v/%v: %v", name, ing.GetNamespace(), ing.GetName(), err)
klog.V(5).ErrorS(err, "error reading Ingress annotation", "name", name, "namespace", ing.GetNamespace(), "ingress", ing.GetName())
}
if val != nil {
@ -211,7 +211,7 @@ func (e Extractor) Extract(ing *networking.Ingress) *Ingress {
err := mergo.MapWithOverwrite(pia, data)
if err != nil {
klog.Errorf("unexpected error merging extracted annotations: %v", err)
klog.ErrorS(err, "unexpected error merging extracted annotations")
}
return pia

8
internal/ingress/annotations/authreq/main.go
View file

@ -171,17 +171,17 @@ func (a authReq) Parse(ing *networking.Ingress) (interface{}, error) {
// Optional Parameters
signIn, err := parser.GetStringAnnotation("auth-signin", ing)
if err != nil {
klog.V(3).Infof("auth-signin annotation is undefined and will not be set")
klog.V(3).InfoS("auth-signin annotation is undefined and will not be set")
}
authSnippet, err := parser.GetStringAnnotation("auth-snippet", ing)
if err != nil {
klog.V(3).Infof("auth-snippet annotation is undefined and will not be set")
klog.V(3).InfoS("auth-snippet annotation is undefined and will not be set")
}
authCacheKey, err := parser.GetStringAnnotation("auth-cache-key", ing)
if err != nil {
klog.V(3).Infof("auth-cache-key annotation is undefined and will not be set")
klog.V(3).InfoS("auth-cache-key annotation is undefined and will not be set")
}
durstr, _ := parser.GetStringAnnotation("auth-cache-duration", ing)
@ -207,7 +207,7 @@ func (a authReq) Parse(ing *networking.Ingress) (interface{}, error) {
proxySetHeaderMap, err := parser.GetStringAnnotation("auth-proxy-set-headers", ing)
if err != nil {
klog.V(3).Infof("auth-set-proxy-headers annotation is undefined and will not be set")
klog.V(3).InfoS("auth-set-proxy-headers annotation is undefined and will not be set")
}
var proxySetHeaders map[string]string

2
internal/ingress/annotations/secureupstream/main.go
View file

@ -42,7 +42,7 @@ func NewParser(r resolver.Resolver) parser.IngressAnnotation {
// rule used to indicate if the upstream servers should use SSL
func (a su) Parse(ing *networking.Ingress) (secure interface{}, err error) {
if ca, _ := parser.GetStringAnnotation("secure-verify-ca-secret", ing); ca != "" {
klog.Errorf("NOTE! secure-verify-ca-secret is not suppored anymore. Please use proxy-ssl-secret instead")
klog.Warningf("NOTE! secure-verify-ca-secret is not suppored anymore. Please use proxy-ssl-secret instead")
}
return
}

16
internal/ingress/annotations/sessionaffinity/main.go
View file

@ -96,40 +96,40 @@ func (a affinity) cookieAffinityParse(ing *networking.Ingress) *Cookie {
cookie.Name, err = parser.GetStringAnnotation(annotationAffinityCookieName, ing)
if err != nil {
klog.V(3).Infof("Ingress %v: No value found in annotation %v. Using the default %v", ing.Name, annotationAffinityCookieName, defaultAffinityCookieName)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieExpires, "default", defaultAffinityCookieName)
cookie.Name = defaultAffinityCookieName
}
cookie.Expires, err = parser.GetStringAnnotation(annotationAffinityCookieExpires, ing)
if err != nil || !affinityCookieExpiresRegex.MatchString(cookie.Expires) {
klog.V(3).Infof("Invalid or no annotation value found in Ingress %v: %v. Ignoring it", ing.Name, annotationAffinityCookieExpires)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieExpires)
cookie.Expires = ""
}
cookie.MaxAge, err = parser.GetStringAnnotation(annotationAffinityCookieMaxAge, ing)
if err != nil || !affinityCookieExpiresRegex.MatchString(cookie.MaxAge) {
klog.V(3).Infof("Invalid or no annotation value found in Ingress %v: %v. Ignoring it", ing.Name, annotationAffinityCookieMaxAge)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieMaxAge)
cookie.MaxAge = ""
}
cookie.Path, err = parser.GetStringAnnotation(annotationAffinityCookiePath, ing)
if err != nil {
klog.V(3).Infof("Invalid or no annotation value found in Ingress %v: %v. Ignoring it", ing.Name, annotationAffinityCookieMaxAge)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieMaxAge)
}
cookie.SameSite, err = parser.GetStringAnnotation(annotationAffinityCookieSameSite, ing)
if err != nil {
klog.V(3).Infof("Invalid or no annotation value found in Ingress %v: %v. Ignoring it", ing.Name, annotationAffinityCookieSameSite)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieSameSite)
}
cookie.ConditionalSameSiteNone, err = parser.GetBoolAnnotation(annotationAffinityCookieConditionalSameSiteNone, ing)
if err != nil {
klog.V(3).Infof("Invalid or no annotation value found in Ingress %v: %v. Ignoring it", ing.Name, annotationAffinityCookieConditionalSameSiteNone)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieConditionalSameSiteNone)
}
cookie.ChangeOnFailure, err = parser.GetBoolAnnotation(annotationAffinityCookieChangeOnFailure, ing)
if err != nil {
klog.V(3).Infof("Invalid or no annotation value found in Ingress %v: %v. Ignoring it", ing.Name, annotationAffinityCookieChangeOnFailure)
klog.V(3).InfoS("Invalid or no annotation value found. Ignoring", "ingress", ing.Name, "annotation", annotationAffinityCookieChangeOnFailure)
}
return cookie
@ -164,7 +164,7 @@ func (a affinity) Parse(ing *networking.Ingress) (interface{}, error) {
case "cookie":
cookie = a.cookieAffinityParse(ing)
default:
klog.V(3).Infof("No default affinity was found for Ingress %v", ing.Name)
klog.V(3).InfoS("No default affinity found", "ingress", ing.Name)
}

15
internal/ingress/controller/controller.go
View file

@ -135,14 +135,14 @@ func (n *NGINXController) syncIngress(interface{}) error {
n.metricCollector.SetSSLExpireTime(servers)
if n.runningConfig.Equal(pcfg) {
klog.V(3).Infof("No configuration change detected, skipping backend reload.")
klog.V(3).Infof("No configuration change detected, skipping backend reload")
return nil
}
n.metricCollector.SetHosts(hosts)
if !n.IsDynamicConfigurationEnough(pcfg) {
klog.Infof("Configuration changes detected, backend reload required.")
klog.InfoS("Configuration changes detected, backend reload required")
hash, _ := hashstructure.Hash(pcfg, &hashstructure.HashOptions{
TagName: "json",
@ -159,7 +159,7 @@ func (n *NGINXController) syncIngress(interface{}) error {
return err
}
klog.Infof("Backend successfully reloaded.")
klog.InfoS("Backend successfully reloaded")
n.metricCollector.ConfigSuccess(hash, true)
n.metricCollector.IncReloadCount()
@ -170,7 +170,7 @@ func (n *NGINXController) syncIngress(interface{}) error {
if isFirstSync {
// For the initial sync it always takes some time for NGINX to start listening
// For large configurations it might take a while so we loop and back off
klog.Info("Initial sync, sleeping for 1 second.")
klog.InfoS("Initial sync, sleeping for 1 second")
time.Sleep(1 * time.Second)
}
@ -1142,7 +1142,7 @@ func (n *NGINXController) createServers(data []*ingress.Ingress,
tlsSecretName := extractTLSSecretName(host, ing, n.store.GetLocalSSLCert)
if tlsSecretName == "" {
klog.V(3).Infof("Host %q is listed in the TLS section but secretName is empty. Using default certificate.", host)
klog.V(3).Infof("Host %q is listed in the TLS section but secretName is empty. Using default certificate", host)
servers[host].SSLCert = n.getDefaultSSLCertificate()
continue
}
@ -1165,13 +1165,12 @@ func (n *NGINXController) createServers(data []*ingress.Ingress,
err = cert.Certificate.VerifyHostname(host)
if err != nil {
klog.Warningf("Unexpected error validating SSL certificate %q for server %q: %v", secrKey, host, err)
klog.Warning("Validating certificate against DNS names. This will be deprecated in a future version.")
klog.Warning("Validating certificate against DNS names. This will be deprecated in a future version")
// check the Common Name field
// https://github.com/golang/go/issues/22922
err := verifyHostname(host, cert.Certificate)
if err != nil {
klog.Warningf("SSL certificate %q does not contain a Common Name or Subject Alternative Name for server %q: %v",
secrKey, host, err)
klog.Warningf("SSL certificate %q does not contain a Common Name or Subject Alternative Name for server %q: %v", secrKey, host, err)
klog.Warningf("Using default certificate")
servers[host].SSLCert = n.getDefaultSSLCertificate()
continue

51
internal/ingress/controller/nginx.go
View file

@ -147,16 +147,12 @@ func NewNGINXController(config *Configuration, mc metric.Collector) *NGINXContro
template, err := ngx_template.NewTemplate(nginx.TemplatePath)
if err != nil {
// this error is different from the rest because it must be clear why nginx is not working
klog.Errorf(`
-------------------------------------------------------------------------------
Error loading new template: %v
-------------------------------------------------------------------------------
`, err)
klog.ErrorS(err, "Error loading new template")
return
}
n.t = template
klog.Info("New NGINX configuration template loaded.")
klog.InfoS("New NGINX configuration template loaded")
n.syncQueue.EnqueueTask(task.GetDummyObject("template-change"))
}
@ -192,7 +188,7 @@ Error loading new template: %v
for _, f := range filesToWatch {
_, err = watch.NewFileWatcher(f, func() {
klog.Infof("File %v changed. Reloading NGINX", f)
klog.InfoS("File changed detected. Reloading NGINX", "path", f)
n.syncQueue.EnqueueTask(task.GetDummyObject("file-change"))
})
if err != nil {
@ -250,7 +246,7 @@ type NGINXController struct {
// Start starts a new NGINX master process running in the foreground.
func (n *NGINXController) Start() {
klog.Info("Starting NGINX Ingress controller")
klog.InfoS("Starting NGINX Ingress controller")
n.store.Run(n.stopCh)
@ -292,7 +288,7 @@ func (n *NGINXController) Start() {
n.setupSSLProxy()
}
klog.Info("Starting NGINX process")
klog.InfoS("Starting NGINX process")
n.start(cmd)
go n.syncQueue.Run(time.Second, n.stopCh)
@ -306,15 +302,16 @@ func (n *NGINXController) Start() {
time.Sleep(5 * time.Minute)
err := cleanTempNginxCfg()
if err != nil {
klog.Infof("Unexpected error removing temporal configuration files: %v", err)
klog.ErrorS(err, "Unexpected error removing temporal configuration files")
}
}
}()
if n.validationWebhookServer != nil {
klog.Infof("Starting validation webhook on %s with keys %s %s", n.validationWebhookServer.Addr, n.cfg.ValidationWebhookCertPath, n.cfg.ValidationWebhookKeyPath)
klog.InfoS("Starting validation webhook", "address", n.validationWebhookServer.Addr,
"certPath", n.cfg.ValidationWebhookCertPath, "keyPath", n.cfg.ValidationWebhookKeyPath)
go func() {
klog.Error(n.validationWebhookServer.ListenAndServeTLS("", ""))
klog.ErrorS(n.validationWebhookServer.ListenAndServeTLS("", ""), "Error listening for TLS connections")
}()
}
@ -337,7 +334,7 @@ func (n *NGINXController) Start() {
}
if evt, ok := event.(store.Event); ok {
klog.V(3).Infof("Event %v received - object %v", evt.Type, evt.Obj)
klog.V(3).InfoS("Event received", "type", evt.Type, "object", evt.Obj)
if evt.Type == store.ConfigurationEvent {
// TODO: is this necessary? Consider removing this special case
n.syncQueue.EnqueueTask(task.GetDummyObject("configmap-change"))
@ -365,7 +362,7 @@ func (n *NGINXController) Stop() error {
return fmt.Errorf("shutdown already in progress")
}
klog.Info("Shutting down controller queues")
klog.InfoS("Shutting down controller queues")
close(n.stopCh)
go n.syncQueue.Shutdown()
if n.syncStatus != nil {
@ -373,7 +370,7 @@ func (n *NGINXController) Stop() error {
}
if n.validationWebhookServer != nil {
klog.Info("Stopping admission controller")
klog.InfoS("Stopping admission controller")
err := n.validationWebhookServer.Close()
if err != nil {
return err
@ -381,7 +378,7 @@ func (n *NGINXController) Stop() error {
}
// send stop signal to NGINX
klog.Info("Stopping NGINX process")
klog.InfoS("Stopping NGINX process")
cmd := n.command.ExecCommand("-s", "quit")
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
@ -394,7 +391,7 @@ func (n *NGINXController) Stop() error {
timer := time.NewTicker(time.Second * 1)
for range timer.C {
if !nginx.IsRunning() {
klog.Info("NGINX process has stopped")
klog.InfoS("NGINX process has stopped")
timer.Stop()
break
}
@ -495,13 +492,13 @@ func (n NGINXController) generateTemplate(cfg ngx_config.Configuration, ingressC
nameHashBucketSize := nginxHashBucketSize(longestName)
if cfg.ServerNameHashBucketSize < nameHashBucketSize {
klog.V(3).Infof("Adjusting ServerNameHashBucketSize variable to %d", nameHashBucketSize)
klog.V(3).InfoS("Adjusting ServerNameHashBucketSize variable", "value", nameHashBucketSize)
cfg.ServerNameHashBucketSize = nameHashBucketSize
}
serverNameHashMaxSize := nextPowerOf2(serverNameBytes)
if cfg.ServerNameHashMaxSize < serverNameHashMaxSize {
klog.V(3).Infof("Adjusting ServerNameHashMaxSize variable to %d", serverNameHashMaxSize)
klog.V(3).InfoS("Adjusting ServerNameHashMaxSize variable", "value", serverNameHashMaxSize)
cfg.ServerNameHashMaxSize = serverNameHashMaxSize
}
@ -509,23 +506,23 @@ func (n NGINXController) generateTemplate(cfg ngx_config.Configuration, ingressC
// the limit of open files is per worker process
// and we leave some room to avoid consuming all the FDs available
wp, err := strconv.Atoi(cfg.WorkerProcesses)
klog.V(3).Infof("Number of worker processes: %d", wp)
klog.V(3).InfoS("Worker processes", "count", wp)
if err != nil {
wp = 1
}
maxOpenFiles := (rlimitMaxNumFiles() / wp) - 1024
klog.V(3).Infof("Maximum number of open file descriptors: %d", maxOpenFiles)
klog.V(3).InfoS("Maximum number of open file descriptors", "value", maxOpenFiles)
if maxOpenFiles < 1024 {
// this means the value of RLIMIT_NOFILE is too low.
maxOpenFiles = 1024
}
klog.V(3).Infof("Adjusting MaxWorkerOpenFiles variable to %d", maxOpenFiles)
klog.V(3).InfoS("Adjusting MaxWorkerOpenFiles variable", "value", maxOpenFiles)
cfg.MaxWorkerOpenFiles = maxOpenFiles
}
if cfg.MaxWorkerConnections == 0 {
maxWorkerConnections := int(float64(cfg.MaxWorkerOpenFiles * 3.0 / 4))
klog.V(3).Infof("Adjusting MaxWorkerConnections variable to %d", maxWorkerConnections)
klog.V(3).InfoS("Adjusting MaxWorkerConnections variable", "value", maxWorkerConnections)
cfg.MaxWorkerConnections = maxWorkerConnections
}
@ -684,7 +681,7 @@ func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {
}
}
klog.Infof("NGINX configuration diff:\n%v", string(diffOutput))
klog.InfoS("NGINX configuration change", "diff", string(diffOutput))
// we do not defer the deletion of temp files in order
// to keep them around for inspection in case of error
@ -735,7 +732,7 @@ func (n *NGINXController) setupSSLProxy() {
sslPort := n.cfg.ListenPorts.HTTPS
proxyPort := n.cfg.ListenPorts.SSLProxy
klog.Info("Starting TLS proxy for SSL Passthrough")
klog.InfoS("Starting TLS proxy for SSL Passthrough")
n.Proxy = &TCPProxy{
Default: &TCPServer{
Hostname: "localhost",
@ -771,7 +768,7 @@ func (n *NGINXController) setupSSLProxy() {
continue
}
klog.V(3).Infof("Handling connection from remote address %s to local %s", conn.RemoteAddr(), conn.LocalAddr())
klog.V(3).InfoS("Handling TCP connection", "remote", conn.RemoteAddr(), "local", conn.LocalAddr())
go n.Proxy.Handle(conn)
}
}()
@ -1146,7 +1143,7 @@ func buildRedirects(servers []*ingress.Server) []*redirect {
continue
}
klog.V(3).Infof("Creating redirect from %q to %q", from, to)
klog.V(3).InfoS("Creating redirect", "from", from, "to", to)
found := false
for _, esrv := range servers {
if esrv.Hostname == from {

6
internal/ingress/controller/status.go
View file

@ -60,7 +60,7 @@ func setupLeaderElection(config *leaderElectionConfig) {
var stopCh chan struct{}
callbacks := leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
klog.V(2).Infof("I am the new leader")
klog.V(2).InfoS("I am the new leader")
stopCh = make(chan struct{})
if config.OnStartedLeading != nil {
@ -68,7 +68,7 @@ func setupLeaderElection(config *leaderElectionConfig) {
}
},
OnStoppedLeading: func() {
klog.V(2).Info("I am not leader anymore")
klog.V(2).InfoS("I am not leader anymore")
close(stopCh)
// cancel the context
@ -81,7 +81,7 @@ func setupLeaderElection(config *leaderElectionConfig) {
}
},
OnNewLeader: func(identity string) {
klog.Infof("new leader elected: %v", identity)
klog.InfoS("New leader elected", "identity", identity)
},
}

10
internal/ingress/controller/store/backend_ssl.go
View file

@ -38,7 +38,7 @@ func (s *k8sStore) syncSecret(key string) {
s.syncSecretMu.Lock()
defer s.syncSecretMu.Unlock()
klog.V(3).Infof("Syncing Secret %q", key)
klog.V(3).InfoS("Syncing Secret", "name", key)
cert, err := s.getPemCertificate(key)
if err != nil {
@ -55,7 +55,7 @@ func (s *k8sStore) syncSecret(key string) {
// no need to update
return
}
klog.Infof("Updating Secret %q in the local store", key)
klog.InfoS("Updating secret in local store", "name", key)
s.sslStore.Update(key, cert)
// this update must trigger an update
// (like an update event from a change in Ingress)
@ -63,7 +63,7 @@ func (s *k8sStore) syncSecret(key string) {
return
}
klog.Infof("Adding Secret %q to the local store", key)
klog.InfoS("Adding secret to local store", "name", key)
s.sslStore.Add(key, cert)
// this update must trigger an update
// (like an update event from a change in Ingress)
@ -142,7 +142,7 @@ func (s *k8sStore) getPemCertificate(secretName string) (*ingress.SSLCert, error
msg += " and CRL"
}
klog.V(3).Info(msg)
klog.V(3).InfoS(msg)
} else if len(ca) > 0 {
sslCert, err = ssl.CreateCACert(ca)
if err != nil {
@ -162,7 +162,7 @@ func (s *k8sStore) getPemCertificate(secretName string) (*ingress.SSLCert, error
}
// makes this secret in 'syncSecret' to be used for Certificate Authentication
// this does not enable Certificate Authentication
klog.V(3).Infof("Configuring Secret %q for TLS authentication", secretName)
klog.V(3).InfoS("Configuring Secret for TLS authentication", "secret", secretName)
} else {
if auth != nil {
return nil, ErrSecretForAuth

34
internal/ingress/controller/store/store.go
View file

@ -287,7 +287,7 @@ func New(
// If we reached here it means the ingress was deleted but its final state is unrecorded.
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
if !ok {
klog.Errorf("couldn't get object from tombstone %#v", obj)
klog.ErrorS(nil, "Error obtaining object from tombstone", "key", obj)
return
}
ing, ok = tombstone.Obj.(*networkingv1beta1.Ingress)
@ -298,13 +298,14 @@ func New(
}
if !class.IsValid(ing) {
klog.Infof("ignoring delete for ingress %v based on annotation %v", ing.Name, class.IngressKey)
return
}
if isCatchAllIngress(ing.Spec) && disableCatchAll {
klog.Infof("ignoring delete for catch-all ingress %v/%v because of --disable-catch-all", ing.Namespace, ing.Name)
klog.InfoS("Ignoring delete for catch-all because of --disable-catch-all", "namespace", ing.Namespace, "ingress", ing.Name)
return
}
recorder.Eventf(ing, corev1.EventTypeNormal, "DELETE", fmt.Sprintf("Ingress %s/%s", ing.Namespace, ing.Name))
store.listers.IngressWithAnnotation.Delete(ing)
@ -323,11 +324,11 @@ func New(
ing, _ := toIngress(obj)
if !class.IsValid(ing) {
a, _ := parser.GetStringAnnotation(class.IngressKey, ing)
klog.Infof("ignoring add for ingress %v based on annotation %v with value %v", ing.Name, class.IngressKey, a)
klog.InfoS("Ignoring add for ingress based on annotation", "namespace", ing.Namespace, "ingress", ing.Name, "annotation", a)
return
}
if isCatchAllIngress(ing.Spec) && disableCatchAll {
klog.Infof("ignoring add for catch-all ingress %v/%v because of --disable-catch-all", ing.Namespace, ing.Name)
klog.InfoS("Ignoring add for catch-all ingress because of --disable-catch-all", "namespace", ing.Namespace, "ingress", ing.Name)
return
}
recorder.Eventf(ing, corev1.EventTypeNormal, "CREATE", fmt.Sprintf("Ingress %s/%s", ing.Namespace, ing.Name))
@ -350,26 +351,26 @@ func New(
validCur := class.IsValid(curIng)
if !validOld && validCur {
if isCatchAllIngress(curIng.Spec) && disableCatchAll {
klog.Infof("ignoring update for catch-all ingress %v/%v because of --disable-catch-all", curIng.Namespace, curIng.Name)
klog.InfoS("ignoring update for catch-all ingress because of --disable-catch-all", "namespace", curIng.Namespace, "ingress", curIng.Name)
return
}
klog.Infof("creating ingress %v based on annotation %v", curIng.Name, class.IngressKey)
klog.InfoS("creating ingress", "namespace", curIng.Namespace, "ingress", curIng.Name, "class", class.IngressKey)
recorder.Eventf(curIng, corev1.EventTypeNormal, "CREATE", fmt.Sprintf("Ingress %s/%s", curIng.Namespace, curIng.Name))
} else if validOld && !validCur {
klog.Infof("removing ingress %v based on annotation %v", curIng.Name, class.IngressKey)
klog.InfoS("removing ingress", "namespace", curIng.Namespace, "ingress", curIng.Name, "class", class.IngressKey)
ingDeleteHandler(old)
return
} else if validCur && !reflect.DeepEqual(old, cur) {
if isCatchAllIngress(curIng.Spec) && disableCatchAll {
klog.Infof("ignoring update for catch-all ingress %v/%v and delete old one because of --disable-catch-all", curIng.Namespace, curIng.Name)
klog.InfoS("ignoring update for catch-all ingress and delete old one because of --disable-catch-all", "namespace", curIng.Namespace, "ingress", curIng.Name)
ingDeleteHandler(old)
return
}
recorder.Eventf(curIng, corev1.EventTypeNormal, "UPDATE", fmt.Sprintf("Ingress %s/%s", curIng.Namespace, curIng.Name))
} else {
klog.V(3).Infof("No changes on ingress %v/%v. Skipping update", curIng.Namespace, curIng.Name)
klog.V(3).InfoS("No changes on ingress. Skipping update", "namespace", curIng.Namespace, "ingress", curIng.Name)
return
}
@ -395,7 +396,7 @@ func New(
// find references in ingresses and update local ssl certs
if ings := store.secretIngressMap.Reference(key); len(ings) > 0 {
klog.Infof("secret %v was added and it is used in ingress annotations. Parsing...", key)
klog.InfoS("Secret was added and it is used in ingress annotations. Parsing", "secret", key)
for _, ingKey := range ings {
ing, err := store.getIngress(ingKey)
if err != nil {
@ -422,11 +423,11 @@ func New(
// find references in ingresses and update local ssl certs
if ings := store.secretIngressMap.Reference(key); len(ings) > 0 {
klog.Infof("secret %v was updated and it is used in ingress annotations. Parsing...", key)
klog.InfoS("secret was updated and it is used in ingress annotations. Parsing", "secret", key)
for _, ingKey := range ings {
ing, err := store.getIngress(ingKey)
if err != nil {
klog.Errorf("could not find Ingress %v in local store", ingKey)
klog.ErrorS(err, "could not find Ingress in local store", "ingress", ingKey)
continue
}
store.syncIngress(ing)
@ -445,12 +446,11 @@ func New(
// If we reached here it means the secret was deleted but its final state is unrecorded.
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
if !ok {
klog.Errorf("couldn't get object from tombstone %#v", obj)
return
}
sec, ok = tombstone.Obj.(*corev1.Secret)
if !ok {
klog.Errorf("Tombstone contained object that is not a Secret: %#v", obj)
return
}
}
@ -461,7 +461,7 @@ func New(
// find references in ingresses
if ings := store.secretIngressMap.Reference(key); len(ings) > 0 {
klog.Infof("secret %v was deleted and it is used in ingress annotations. Parsing...", key)
klog.InfoS("secret was deleted and it is used in ingress annotations. Parsing", "secret", key)
for _, ingKey := range ings {
ing, err := store.getIngress(ingKey)
if err != nil {
@ -859,7 +859,7 @@ func (s *k8sStore) setConfig(cmap *corev1.ConfigMap) {
s.backendConfig = ngx_template.ReadConfig(cmap.Data)
if s.backendConfig.UseGeoIP2 && !nginx.GeoLite2DBExists() {
klog.Warning("The GeoIP2 feature is enabled but the databases are missing. Disabling.")
klog.Warning("The GeoIP2 feature is enabled but the databases are missing. Disabling")
s.backendConfig.UseGeoIP2 = false
}

10
internal/ingress/controller/tcp.go
View file

@ -63,19 +63,19 @@ func (p *TCPProxy) Handle(conn net.Conn) {
length, err := conn.Read(data)
if err != nil {
klog.V(4).Infof("Error reading the first 4k of the connection: %v", err)
klog.V(4).ErrorS(err, "Error reading the first 4k of the connection")
return
}
proxy := p.Default
hostname, err := parser.GetHostname(data[:])
if err == nil {
klog.V(4).Infof("Parsed hostname from TLS Client Hello: %s", hostname)
klog.V(4).InfoS("TLS Client Hello", "host", hostname)
proxy = p.Get(hostname)
}
if proxy == nil {
klog.V(4).Info("There is no configured proxy for SSL connections.")
klog.V(4).InfoS("There is no configured proxy for SSL connections.")
return
}
@ -97,11 +97,11 @@ func (p *TCPProxy) Handle(conn net.Conn) {
protocol = "TCP6"
}
proxyProtocolHeader := fmt.Sprintf("PROXY %s %s %s %d %d\r\n", protocol, remoteAddr.IP.String(), localAddr.IP.String(), remoteAddr.Port, localAddr.Port)
klog.V(4).Infof("Writing Proxy Protocol header: %s", proxyProtocolHeader)
klog.V(4).InfoS("Writing Proxy Protocol", "header", proxyProtocolHeader)
_, err = fmt.Fprintf(clientConn, proxyProtocolHeader)
}
if err != nil {
klog.Errorf("Error writing Proxy Protocol header: %v", err)
klog.ErrorS(err, "Error writing Proxy Protocol header")
clientConn.Close()
} else {
_, err = clientConn.Write(data[:length])

2
internal/ingress/controller/template/template.go
View file

@ -100,7 +100,7 @@ func (t *Template) Write(conf config.TemplateConfig) ([]byte, error) {
if err != nil {
klog.Errorf("unexpected error: %v", err)
}
klog.Infof("NGINX configuration: %v", string(b))
klog.InfoS("NGINX", "configuration", string(b))
}
err := t.tmpl.Execute(tmplBuf, conf)

5
internal/ingress/controller/util.go
View file

@ -54,7 +54,7 @@ func upstreamName(namespace string, service string, port intstr.IntOrString) str
func sysctlSomaxconn() int {
maxConns, err := sysctl.New().GetSysctl("net/core/somaxconn")
if err != nil || maxConns < 512 {
klog.V(3).Infof("net.core.somaxconn=%v (using system default)", maxConns)
klog.V(3).InfoS("Using default net.core.somaxconn", "value", maxConns)
return 511
}
@ -66,10 +66,9 @@ func rlimitMaxNumFiles() int {
var rLimit syscall.Rlimit
err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit)
if err != nil {
klog.Errorf("Error reading system maximum number of open file descriptors (RLIMIT_NOFILE): %v", err)
klog.ErrorS(err, "Error reading system maximum number of open file descriptors (RLIMIT_NOFILE)")
return 0
}
klog.V(2).Infof("rlimit.max=%v", rLimit.Max)
return int(rLimit.Max)
}

6
internal/ingress/metric/collectors/controller.go
View file

@ -253,7 +253,7 @@ func (cm *Controller) RemoveAllSSLExpireMetrics(registry prometheus.Gatherer) {
func (cm *Controller) removeSSLExpireMetrics(onlyDefinedHosts bool, hosts []string, registry prometheus.Gatherer) {
mfs, err := registry.Gather()
if err != nil {
klog.Errorf("Error gathering metrics: %v", err)
klog.ErrorS(err, "Error gathering metrics")
return
}
@ -283,10 +283,10 @@ func (cm *Controller) removeSSLExpireMetrics(onlyDefinedHosts bool, hosts []stri
continue
}
klog.V(2).Infof("Removing prometheus metric from gauge %v for host %v", metricName, host)
klog.V(2).InfoS("Removing prometheus metric", "gauge", metricName, "host", host)
removed := cm.sslExpireTime.Delete(labels)
if !removed {
klog.V(2).Infof("metric %v for host %v with labels not removed: %v", metricName, host, labels)
klog.V(2).InfoS("metric removed", "metric", metricName, "host", host, "labels", labels)
}
}
}

2
internal/ingress/metric/collectors/nginx_status.go
View file

@ -165,7 +165,7 @@ func parse(data string) *basicStatus {
// nginxStatusCollector scrape the nginx status
func (p nginxStatusCollector) scrape(ch chan<- prometheus.Metric) {
klog.V(3).Infof("start scraping socket: %v", nginx.StatusPath)
klog.V(3).InfoS("starting scraping socket", "path", nginx.StatusPath)
status, data, err := nginx.NewGetStatusRequest(nginx.StatusPath)
if err != nil {
log.Printf("%v", err)

29
internal/ingress/metric/collectors/socket.go
View file

@ -219,19 +219,19 @@ func NewSocketCollector(pod, namespace, class string, metricsPerHost bool) (*Soc
}
func (sc *SocketCollector) handleMessage(msg []byte) {
klog.V(5).Infof("msg: %v", string(msg))
klog.V(5).InfoS("Metric", "message", string(msg))
// Unmarshal bytes
var statsBatch []socketData
err := jsoniter.ConfigCompatibleWithStandardLibrary.Unmarshal(msg, &statsBatch)
if err != nil {
klog.Errorf("Unexpected error deserializing JSON payload: %v. Payload:\n%v", err, string(msg))
klog.ErrorS(err, "Unexpected error deserializing JSON", "payload", string(msg))
return
}
for _, stats := range statsBatch {
if sc.metricsPerHost && !sc.hosts.Has(stats.Host) {
klog.V(3).Infof("Skipping metric for host %v that is not being served", stats.Host)
klog.V(3).InfoS("Skipping metric for host not being served", "host", stats.Host)
continue
}
@ -263,7 +263,7 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
requestsMetric, err := sc.requests.GetMetricWith(collectorLabels)
if err != nil {
klog.Errorf("Error fetching requests metric: %v", err)
klog.ErrorS(err, "Error fetching requests metric")
} else {
requestsMetric.Inc()
}
@ -271,7 +271,7 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
if stats.Latency != -1 {
latencyMetric, err := sc.upstreamLatency.GetMetricWith(latencyLabels)
if err != nil {
klog.Errorf("Error fetching latency metric: %v", err)
klog.ErrorS(err, "Error fetching latency metric")
} else {
latencyMetric.Observe(stats.Latency)
}
@ -280,7 +280,7 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
if stats.RequestTime != -1 {
requestTimeMetric, err := sc.requestTime.GetMetricWith(requestLabels)
if err != nil {
klog.Errorf("Error fetching request duration metric: %v", err)
klog.ErrorS(err, "Error fetching request duration metric")
} else {
requestTimeMetric.Observe(stats.RequestTime)
}
@ -289,7 +289,7 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
if stats.RequestLength != -1 {
requestLengthMetric, err := sc.requestLength.GetMetricWith(requestLabels)
if err != nil {
klog.Errorf("Error fetching request length metric: %v", err)
klog.ErrorS(err, "Error fetching request length metric")
} else {
requestLengthMetric.Observe(stats.RequestLength)
}
@ -298,7 +298,7 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
if stats.ResponseTime != -1 {
responseTimeMetric, err := sc.responseTime.GetMetricWith(requestLabels)
if err != nil {
klog.Errorf("Error fetching upstream response time metric: %v", err)
klog.ErrorS(err, "Error fetching upstream response time metric")
} else {
responseTimeMetric.Observe(stats.ResponseTime)
}
@ -307,14 +307,14 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
if stats.ResponseLength != -1 {
bytesSentMetric, err := sc.bytesSent.GetMetricWith(requestLabels)
if err != nil {
klog.Errorf("Error fetching bytes sent metric: %v", err)
klog.ErrorS(err, "Error fetching bytes sent metric")
} else {
bytesSentMetric.Observe(stats.ResponseLength)
}
responseSizeMetric, err := sc.responseLength.GetMetricWith(requestLabels)
if err != nil {
klog.Errorf("Error fetching bytes sent metric: %v", err)
klog.ErrorS(err, "Error fetching bytes sent metric")
} else {
responseSizeMetric.Observe(stats.ResponseLength)
}
@ -345,12 +345,12 @@ func (sc *SocketCollector) Stop() {
func (sc *SocketCollector) RemoveMetrics(ingresses []string, registry prometheus.Gatherer) {
mfs, err := registry.Gather()
if err != nil {
klog.Errorf("Error gathering metrics: %v", err)
klog.ErrorS(err, "Error gathering metrics: %v")
return
}
// 1. remove metrics of removed ingresses
klog.V(2).Infof("removing ingresses %v from metrics", ingresses)
klog.V(2).InfoS("removing metrics", "ingresses", ingresses)
for _, mf := range mfs {
metricName := mf.GetName()
metric, ok := sc.metricMapping[metricName]
@ -388,7 +388,7 @@ func (sc *SocketCollector) RemoveMetrics(ingresses []string, registry prometheus
if ok {
removed := h.Delete(labels)
if !removed {
klog.V(2).Infof("metric %v for ingress %v with labels not removed: %v", metricName, ingKey, labels)
klog.V(2).InfoS("metric not removed", "name", metricName, "ingress", ingKey, "labels", labels)
}
}
@ -396,12 +396,11 @@ func (sc *SocketCollector) RemoveMetrics(ingresses []string, registry prometheus
if ok {
removed := s.Delete(labels)
if !removed {
klog.V(2).Infof("metric %v for ingress %v with labels not removed: %v", metricName, ingKey, labels)
klog.V(2).InfoS("metric not removed", "name", metricName, "ingress", ingKey, "labels", labels)
}
}
}
}
}
// Describe implements prometheus.Collector

2
internal/ingress/metric/main.go
View file

@ -160,7 +160,7 @@ func (c *collector) SetSSLExpireTime(servers []*ingress.Server) {
return
}
klog.V(2).Infof("Updating ssl expiration metrics.")
klog.V(2).InfoS("Updating ssl expiration metrics")
c.ingressController.SetSSLExpireTime(servers)
}

16
internal/ingress/status/status.go
View file

@ -112,32 +112,30 @@ func (s statusSync) Shutdown() {
return
}
klog.Info("updating status of Ingress rules (remove)")
addrs, err := s.runningAddresses()
if err != nil {
klog.Errorf("error obtaining running IPs: %v", addrs)
klog.ErrorS(err, "error obtaining running IP address")
return
}
if len(addrs) > 1 {
// leave the job to the next leader
klog.Infof("leaving status update for next leader (%v)", len(addrs))
klog.InfoS("leaving status update for next leader")
return
}
if s.isRunningMultiplePods() {
klog.V(2).Infof("skipping Ingress status update (multiple pods running - another one will be elected as master)")
klog.V(2).InfoS("skipping Ingress status update (multiple pods running - another one will be elected as master)")
return
}
klog.Infof("removing address from ingress status (%v)", addrs)
klog.InfoS("removing value from ingress status", "address", addrs)
s.updateStatus([]apiv1.LoadBalancerIngress{})
}
func (s *statusSync) sync(key interface{}) error {
if s.syncQueue.IsShuttingDown() {
klog.V(2).Infof("skipping Ingress status update (shutting down in progress)")
klog.V(2).InfoS("skipping Ingress status update (shutting down in progress)")
return nil
}
@ -252,7 +250,7 @@ func (s *statusSync) updateStatus(newIngressPoint []apiv1.LoadBalancerIngress) {
curIPs := ing.Status.LoadBalancer.Ingress
sort.SliceStable(curIPs, lessLoadBalancerIngress(curIPs))
if ingressSliceEqual(curIPs, newIngressPoint) {
klog.V(3).Infof("skipping update of Ingress %v/%v (no change)", ing.Namespace, ing.Name)
klog.V(3).InfoS("skipping update of Ingress (no change)", "namespace", ing.Namespace, "ingress", ing.Name)
continue
}
@ -276,7 +274,7 @@ func runUpdate(ing *ingress.Ingress, status []apiv1.LoadBalancerIngress,
return nil, errors.Wrap(err, fmt.Sprintf("unexpected error searching Ingress %v/%v", ing.Namespace, ing.Name))
}
klog.Infof("updating Ingress %v/%v status from %v to %v", currIng.Namespace, currIng.Name, currIng.Status.LoadBalancer.Ingress, status)
klog.InfoS("updating Ingress status", "namespace", currIng.Namespace, "ingress", currIng.Name, "currentValue", currIng.Status.LoadBalancer.Ingress, "newValue", status)
currIng.Status.LoadBalancer.Ingress = status
_, err = ingClient.UpdateStatus(context.TODO(), currIng, metav1.UpdateOptions{})
if err != nil {

4
internal/k8s/main.go
View file

@ -46,7 +46,7 @@ func ParseNameNS(input string) (string, string, error) {
func GetNodeIPOrName(kubeClient clientset.Interface, name string, useInternalIP bool) string {
node, err := kubeClient.CoreV1().Nodes().Get(context.TODO(), name, metav1.GetOptions{})
if err != nil {
klog.Errorf("Error getting node %v: %v", name, err)
klog.ErrorS(err, "Error getting node", "name", name)
return ""
}
@ -155,7 +155,7 @@ func NetworkingIngressAvailable(client clientset.Interface) (bool, bool) {
runningVersion, err := version.ParseGeneric(serverVersion.String())
if err != nil {
klog.Errorf("unexpected error parsing running Kubernetes version: %v", err)
klog.ErrorS(err, "unexpected error parsing running Kubernetes version")
return false, false
}

2
internal/net/dns/dns.go
View file

@ -53,6 +53,6 @@ func GetSystemNameServers() ([]net.IP, error) {
}
}
klog.V(3).Infof("nameservers IP address/es to use: %v", nameservers)
klog.V(3).InfoS("Nameservers", "hosts", nameservers)
return nameservers, nil
}

10
internal/net/ssl/ssl.go
View file

@ -73,7 +73,7 @@ func CreateSSLCert(cert, key []byte, uid string) (*ingress.SSLCert, error) {
if ngx_config.EnableSSLChainCompletion {
data, err := fullChainCert(cert)
if err != nil {
klog.Errorf("Error generating certificate chain for Secret: %v", err)
klog.ErrorS(err, "Error generating certificate chain for Secret")
} else {
pemCertBuffer.Reset()
pemCertBuffer.Write(data)
@ -109,7 +109,7 @@ func CreateSSLCert(cert, key []byte, uid string) (*ingress.SSLCert, error) {
}
if len(pemCert.Extensions) > 0 {
klog.V(3).Info("parsing ssl certificate extensions")
klog.V(3).InfoS("parsing ssl certificate extensions")
for _, ext := range getExtension(pemCert, oidExtensionSubjectAltName) {
dns, _, _, err := parseSANExtension(ext.Value)
if err != nil {
@ -257,7 +257,7 @@ func ConfigureCACert(name string, ca []byte, sslCert *ingress.SSLCert) error {
sslCert.CAFileName = fileName
klog.V(3).Infof("Created CA Certificate for Authentication: %v", fileName)
klog.V(3).InfoS("Created CA Certificate for Authentication", "path", fileName)
return nil
}
@ -334,7 +334,7 @@ func AddOrUpdateDHParam(name string, dh []byte) (string, error) {
tempPemFile, err := ioutil.TempFile(file.DefaultSSLDirectory, pemName)
klog.V(3).Infof("Creating temp file %v for DH param: %v", tempPemFile.Name(), pemName)
klog.V(3).InfoS("Creating temporal file for DH", "path", tempPemFile.Name(), "name", pemName)
if err != nil {
return "", fmt.Errorf("could not create temp pem file %v: %v", pemFileName, err)
}
@ -525,7 +525,7 @@ func (tl *TLSListener) TLSConfig() *tls.Config {
}
func (tl *TLSListener) load() {
klog.Infof("loading tls certificate from certificate path %s and key path %s", tl.certificatePath, tl.keyPath)
klog.InfoS("loading tls certificate", "path", tl.certificatePath, "key", tl.keyPath)
certBytes, err := ioutil.ReadFile(tl.certificatePath)
if err != nil {
tl.certificate = nil

2
internal/nginx/main.go
View file

@ -151,7 +151,7 @@ func Version() string {
cmd := exec.Command("nginx", flag)
out, err := cmd.CombinedOutput()
if err != nil {
klog.Errorf("unexpected error obtaining NGINX version: %v", err)
klog.ErrorS(err, "unexpected error obtaining NGINX version")
return "N/A"
}

12
internal/task/queue.go
View file

@ -75,7 +75,7 @@ func (t *Queue) EnqueueSkippableTask(obj interface{}) {
// enqueue enqueues ns/name of the given api object in the task queue.
func (t *Queue) enqueue(obj interface{}, skippable bool) {
if t.IsShuttingDown() {
klog.Errorf("queue has been shutdown, failed to enqueue: %v", obj)
klog.ErrorS(nil, "queue has been shutdown, failed to enqueue", "key", obj)
return
}
@ -84,10 +84,10 @@ func (t *Queue) enqueue(obj interface{}, skippable bool) {
// make sure the timestamp is bigger than lastSync
ts = time.Now().Add(24 * time.Hour).UnixNano()
}
klog.V(3).Infof("queuing item %v", obj)
klog.V(3).InfoS("queuing", "item", obj)
key, err := t.fn(obj)
if err != nil {
klog.Errorf("%v", err)
klog.ErrorS(err, "creating object key", "item", obj)
return
}
t.queue.Add(Element{
@ -119,15 +119,15 @@ func (t *Queue) worker() {
item := key.(Element)
if t.lastSync > item.Timestamp {
klog.V(3).Infof("skipping %v sync (%v > %v)", item.Key, t.lastSync, item.Timestamp)
klog.V(3).InfoS("skipping sync", "key", item.Key, "last", t.lastSync, "now", item.Timestamp)
t.queue.Forget(key)
t.queue.Done(key)
continue
}
klog.V(3).Infof("syncing %v", item.Key)
klog.V(3).InfoS("syncing", "key", item.Key)
if err := t.sync(key); err != nil {
klog.Warningf("requeuing %v, err %v", item.Key, err)
klog.ErrorS(err, "requeuing", "key", item.Key)
t.queue.AddRateLimited(Element{
Key: item.Key,
Timestamp: time.Now().UnixNano(),