From 45b2b20fd9c08b39cdebf9d9cba9baf41b38d1d1 Mon Sep 17 00:00:00 2001 From: Chen Shou Date: Fri, 17 Jan 2020 23:56:26 +0000 Subject: [PATCH 1/3] Make broker filter/ingress use a PrometheusPort other than 9090 --- cmd/broker/filter/main.go | 19 +++++++++++++++++-- cmd/broker/ingress/main.go | 17 ++++++++++++++++- 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/cmd/broker/filter/main.go b/cmd/broker/filter/main.go index f45d28c6b4d..df0a4a52636 100644 --- a/cmd/broker/filter/main.go +++ b/cmd/broker/filter/main.go @@ -39,6 +39,7 @@ import ( "knative.dev/pkg/injection/sharedmain" + corev1 "k8s.io/api/core/v1" eventingv1alpha1 "knative.dev/eventing/pkg/client/clientset/versioned" eventinginformers "knative.dev/eventing/pkg/client/informers/externalversions" ) @@ -49,7 +50,8 @@ var ( ) const ( - component = "broker_filter" + defaultMerticsPort = 9092 + component = "broker_filter" ) type envConfig struct { @@ -103,7 +105,7 @@ func main() { // Watch the logging config map and dynamically update logging levels. configMapWatcher := configmap.NewInformedWatcher(kubeClient, system.Namespace()) // Watch the observability config map and dynamically update metrics exporter. - configMapWatcher.Watch(metrics.ConfigMapName(), metrics.UpdateExporterFromConfigMap(component, sl)) + configMapWatcher.Watch(metrics.ConfigMapName(), updateExporterFromConfigMap(component, sl)) // TODO change the component name to broker once Stackdriver metrics are approved. // Watch the observability config map and dynamically update request logs. configMapWatcher.Watch(logging.ConfigMapName(), logging.UpdateLevelFromConfigMap(sl, atomicLevel, component)) @@ -146,6 +148,19 @@ func main() { logger.Info("Exiting...") } +// Same as metrics.UpdateExporterFromConfigMap except the PrometheusPort. +func updateExporterFromConfigMap(component string, logger *zap.SugaredLogger) func(configMap *corev1.ConfigMap) { + domain := metrics.Domain() + return func(configMap *corev1.ConfigMap) { + metrics.UpdateExporter(metrics.ExporterOptions{ + Domain: domain, + Component: component, + ConfigMap: configMap.Data, + PrometheusPort: defaultMerticsPort, + }, logger) + } +} + func flush(logger *zap.SugaredLogger) { _ = logger.Sync() metrics.FlushExporter() diff --git a/cmd/broker/ingress/main.go b/cmd/broker/ingress/main.go index 0517f47c29a..87bdbea604b 100644 --- a/cmd/broker/ingress/main.go +++ b/cmd/broker/ingress/main.go @@ -31,6 +31,7 @@ import ( "go.opencensus.io/stats/view" "go.uber.org/zap" + corev1 "k8s.io/api/core/v1" "knative.dev/eventing/pkg/broker" "knative.dev/eventing/pkg/broker/ingress" "knative.dev/eventing/pkg/kncloudevents" @@ -62,6 +63,7 @@ const ( defaultMaxIdleConnections = 1000 defaultMaxIdleConnectionsPerHost = 1000 defaultTTL int32 = 255 + defaultMerticsPort = 9092 component = "broker_ingress" ) @@ -120,7 +122,7 @@ func main() { // Watch the logging config map and dynamically update logging levels. configMapWatcher := configmap.NewInformedWatcher(kubeclient.Get(ctx), system.Namespace()) // Watch the observability config map and dynamically update metrics exporter. - configMapWatcher.Watch(metrics.ConfigMapName(), metrics.UpdateExporterFromConfigMap(component, sl)) + configMapWatcher.Watch(metrics.ConfigMapName(), updateExporterFromConfigMap(component, sl)) // TODO change the component name to broker once Stackdriver metrics are approved. // Watch the observability config map and dynamically update request logs. configMapWatcher.Watch(logging.ConfigMapName(), logging.UpdateLevelFromConfigMap(sl, atomicLevel, component)) @@ -185,6 +187,19 @@ func main() { logger.Info("Exiting...") } +// Same as metrics.UpdateExporterFromConfigMap except the PrometheusPort. +func updateExporterFromConfigMap(component string, logger *zap.SugaredLogger) func(configMap *corev1.ConfigMap) { + domain := metrics.Domain() + return func(configMap *corev1.ConfigMap) { + metrics.UpdateExporter(metrics.ExporterOptions{ + Domain: domain, + Component: component, + ConfigMap: configMap.Data, + PrometheusPort: defaultMerticsPort, + }, logger) + } +} + func flush(logger *zap.SugaredLogger) { _ = logger.Sync() metrics.FlushExporter() From dddee099f24dceb85a291f280bc184c7e7f7d9ad Mon Sep 17 00:00:00 2001 From: Chen Shou Date: Tue, 21 Jan 2020 20:42:37 +0000 Subject: [PATCH 2/3] correct typo --- cmd/broker/filter/main.go | 4 ++-- cmd/broker/ingress/main.go | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/cmd/broker/filter/main.go b/cmd/broker/filter/main.go index df0a4a52636..cf7c4116f4b 100644 --- a/cmd/broker/filter/main.go +++ b/cmd/broker/filter/main.go @@ -50,7 +50,7 @@ var ( ) const ( - defaultMerticsPort = 9092 + defaultMetricsPort = 9092 component = "broker_filter" ) @@ -156,7 +156,7 @@ func updateExporterFromConfigMap(component string, logger *zap.SugaredLogger) fu Domain: domain, Component: component, ConfigMap: configMap.Data, - PrometheusPort: defaultMerticsPort, + PrometheusPort: defaultMetricsPort, }, logger) } } diff --git a/cmd/broker/ingress/main.go b/cmd/broker/ingress/main.go index 87bdbea604b..fd9fa0133eb 100644 --- a/cmd/broker/ingress/main.go +++ b/cmd/broker/ingress/main.go @@ -63,7 +63,7 @@ const ( defaultMaxIdleConnections = 1000 defaultMaxIdleConnectionsPerHost = 1000 defaultTTL int32 = 255 - defaultMerticsPort = 9092 + defaultMetricsPort = 9092 component = "broker_ingress" ) @@ -195,7 +195,7 @@ func updateExporterFromConfigMap(component string, logger *zap.SugaredLogger) fu Domain: domain, Component: component, ConfigMap: configMap.Data, - PrometheusPort: defaultMerticsPort, + PrometheusPort: defaultMetricsPort, }, logger) } } From 2277c35e1833e0aa2ed2e49e06df9489a32e0918 Mon Sep 17 00:00:00 2001 From: Chen Shou Date: Mon, 27 Jan 2020 18:30:05 +0000 Subject: [PATCH 3/3] used new metrics exporter function from pkg --- Gopkg.lock | 4 ++-- cmd/broker/filter/main.go | 23 +++++++----------- cmd/broker/ingress/main.go | 23 +++++++----------- vendor/knative.dev/pkg/Gopkg.lock | 2 ++ vendor/knative.dev/pkg/reconciler/retry.go | 28 ++++++++++++++++++++++ 5 files changed, 48 insertions(+), 32 deletions(-) create mode 100644 vendor/knative.dev/pkg/reconciler/retry.go diff --git a/Gopkg.lock b/Gopkg.lock index 5b646e9ec3c..2f8d804cf71 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1297,7 +1297,7 @@ [[projects]] branch = "master" - digest = "1:92b7055a014502a4b66d16ddc846de570f2d42ebdf9e6dfdc1b6eb2efc573294" + digest = "1:ca4d1d28d6bfcfa603954856eec0636828da1256ecafb3c650af05f64cfb0b6c" name = "knative.dev/pkg" packages = [ "apis", @@ -1399,7 +1399,7 @@ "webhook/resourcesemantics/validation", ] pruneopts = "T" - revision = "5ff923b836abcbeb6ac3ff2784db0d5b5786ade2" + revision = "4252f595d1ff347f0396166a0bd688a14463581a" [[projects]] branch = "master" diff --git a/cmd/broker/filter/main.go b/cmd/broker/filter/main.go index cf7c4116f4b..f0ede0c40ac 100644 --- a/cmd/broker/filter/main.go +++ b/cmd/broker/filter/main.go @@ -39,7 +39,6 @@ import ( "knative.dev/pkg/injection/sharedmain" - corev1 "k8s.io/api/core/v1" eventingv1alpha1 "knative.dev/eventing/pkg/client/clientset/versioned" eventinginformers "knative.dev/eventing/pkg/client/informers/externalversions" ) @@ -105,7 +104,14 @@ func main() { // Watch the logging config map and dynamically update logging levels. configMapWatcher := configmap.NewInformedWatcher(kubeClient, system.Namespace()) // Watch the observability config map and dynamically update metrics exporter. - configMapWatcher.Watch(metrics.ConfigMapName(), updateExporterFromConfigMap(component, sl)) + updateFunc, err := metrics.UpdateExporterFromConfigMapWithOpts(metrics.ExporterOptions{ + Component: component, + PrometheusPort: defaultMetricsPort, + }, sl) + if err != nil { + logger.Fatal("Failed to create metrics exporter update function", zap.Error(err)) + } + configMapWatcher.Watch(metrics.ConfigMapName(), updateFunc) // TODO change the component name to broker once Stackdriver metrics are approved. // Watch the observability config map and dynamically update request logs. configMapWatcher.Watch(logging.ConfigMapName(), logging.UpdateLevelFromConfigMap(sl, atomicLevel, component)) @@ -148,19 +154,6 @@ func main() { logger.Info("Exiting...") } -// Same as metrics.UpdateExporterFromConfigMap except the PrometheusPort. -func updateExporterFromConfigMap(component string, logger *zap.SugaredLogger) func(configMap *corev1.ConfigMap) { - domain := metrics.Domain() - return func(configMap *corev1.ConfigMap) { - metrics.UpdateExporter(metrics.ExporterOptions{ - Domain: domain, - Component: component, - ConfigMap: configMap.Data, - PrometheusPort: defaultMetricsPort, - }, logger) - } -} - func flush(logger *zap.SugaredLogger) { _ = logger.Sync() metrics.FlushExporter() diff --git a/cmd/broker/ingress/main.go b/cmd/broker/ingress/main.go index fd9fa0133eb..ea969a1cbee 100644 --- a/cmd/broker/ingress/main.go +++ b/cmd/broker/ingress/main.go @@ -31,7 +31,6 @@ import ( "go.opencensus.io/stats/view" "go.uber.org/zap" - corev1 "k8s.io/api/core/v1" "knative.dev/eventing/pkg/broker" "knative.dev/eventing/pkg/broker/ingress" "knative.dev/eventing/pkg/kncloudevents" @@ -122,7 +121,14 @@ func main() { // Watch the logging config map and dynamically update logging levels. configMapWatcher := configmap.NewInformedWatcher(kubeclient.Get(ctx), system.Namespace()) // Watch the observability config map and dynamically update metrics exporter. - configMapWatcher.Watch(metrics.ConfigMapName(), updateExporterFromConfigMap(component, sl)) + updateFunc, err := metrics.UpdateExporterFromConfigMapWithOpts(metrics.ExporterOptions{ + Component: component, + PrometheusPort: defaultMetricsPort, + }, sl) + if err != nil { + logger.Fatal("Failed to create metrics exporter update function", zap.Error(err)) + } + configMapWatcher.Watch(metrics.ConfigMapName(), updateFunc) // TODO change the component name to broker once Stackdriver metrics are approved. // Watch the observability config map and dynamically update request logs. configMapWatcher.Watch(logging.ConfigMapName(), logging.UpdateLevelFromConfigMap(sl, atomicLevel, component)) @@ -187,19 +193,6 @@ func main() { logger.Info("Exiting...") } -// Same as metrics.UpdateExporterFromConfigMap except the PrometheusPort. -func updateExporterFromConfigMap(component string, logger *zap.SugaredLogger) func(configMap *corev1.ConfigMap) { - domain := metrics.Domain() - return func(configMap *corev1.ConfigMap) { - metrics.UpdateExporter(metrics.ExporterOptions{ - Domain: domain, - Component: component, - ConfigMap: configMap.Data, - PrometheusPort: defaultMetricsPort, - }, logger) - } -} - func flush(logger *zap.SugaredLogger) { _ = logger.Sync() metrics.FlushExporter() diff --git a/vendor/knative.dev/pkg/Gopkg.lock b/vendor/knative.dev/pkg/Gopkg.lock index 258660c508c..6e6c1293da5 100644 --- a/vendor/knative.dev/pkg/Gopkg.lock +++ b/vendor/knative.dev/pkg/Gopkg.lock @@ -1393,6 +1393,7 @@ "k8s.io/api/admissionregistration/v1beta1", "k8s.io/api/apps/v1", "k8s.io/api/authentication/v1", + "k8s.io/api/autoscaling/v1", "k8s.io/api/autoscaling/v2beta1", "k8s.io/api/batch/v1", "k8s.io/api/core/v1", @@ -1449,6 +1450,7 @@ "k8s.io/client-go/tools/clientcmd", "k8s.io/client-go/tools/metrics", "k8s.io/client-go/tools/record", + "k8s.io/client-go/util/retry", "k8s.io/client-go/util/workqueue", "k8s.io/code-generator/cmd/client-gen", "k8s.io/code-generator/cmd/client-gen/generators/util", diff --git a/vendor/knative.dev/pkg/reconciler/retry.go b/vendor/knative.dev/pkg/reconciler/retry.go new file mode 100644 index 00000000000..63c2af24563 --- /dev/null +++ b/vendor/knative.dev/pkg/reconciler/retry.go @@ -0,0 +1,28 @@ +/* +Copyright 2020 The Knative Authors +Licensed under the Apache License, Veroute.on 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package reconciler + +import ( + "k8s.io/client-go/util/retry" +) + +// RetryUpdateConflicts retries the inner function if it returns conflict errors. +// This can be used to retry status updates without constantly reenqueuing keys. +func RetryUpdateConflicts(updater func(int) error) error { + attempts := 0 + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + err := updater(attempts) + attempts++ + return err + }) +}