Instrument instances types

go.mod

@@ -12,16 +12,22 @@ require (
 	github.com/opentracing/opentracing-go v1.1.0
 	github.com/operator-framework/operator-sdk v0.18.2
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/common v0.9.1
-	github.com/sirupsen/logrus v1.5.0
+	github.com/prometheus/client_golang v1.10.0
+	github.com/prometheus/common v0.18.0
+	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/cobra v1.0.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.7.1
 	github.com/stretchr/testify v1.7.0
 	github.com/uber/jaeger-client-go v2.20.1+incompatible
 	go.opentelemetry.io/otel v0.20.0
+	go.opentelemetry.io/otel/exporters/metric/prometheus v0.20.0
 	go.opentelemetry.io/otel/exporters/trace/jaeger v0.20.0
+	go.opentelemetry.io/otel/metric v0.20.0
+	go.opentelemetry.io/otel/oteltest v0.20.0
 	go.opentelemetry.io/otel/sdk v0.20.0
+	go.opentelemetry.io/otel/sdk/export/metric v0.20.0
+	go.opentelemetry.io/otel/sdk/metric v0.20.0
 	go.opentelemetry.io/otel/trace v0.20.0
 	google.golang.org/grpc v1.32.0 // indirect
 	k8s.io/api v0.20.7

pkg/cmd/start/bootstrap.go

@@ -32,6 +32,7 @@ import (
 	"github.com/jaegertracing/jaeger-operator/pkg/apis"
 	v1 "github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
 	"github.com/jaegertracing/jaeger-operator/pkg/controller"
+	opmetrics "github.com/jaegertracing/jaeger-operator/pkg/metrics"
 	"github.com/jaegertracing/jaeger-operator/pkg/tracing"
 	"github.com/jaegertracing/jaeger-operator/pkg/upgrade"
 	"github.com/jaegertracing/jaeger-operator/pkg/version"
@@ -88,7 +89,10 @@ func bootstrap(ctx context.Context) manager.Manager {
 	serveCRMetrics(ctx, cfg, namespace)
 	createMetricsService(ctx, cfg, namespace)
 	detectOAuthProxyImageStream(ctx, mgr)
-
+	err = opmetrics.Bootstrap(ctx, namespace, mgr.GetClient())
+	if err != nil {
+		log.WithError(err).Error("failed to initialize metrics")
+	}
 	return mgr
 }
 

pkg/metrics/bootstrap.go

@@ -0,0 +1,57 @@
+package metrics
+
+import (
+	"context"
+
+	prometheusclient "github.com/prometheus/client_golang/prometheus"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/metric/prometheus"
+	"go.opentelemetry.io/otel/metric/global"
+	export "go.opentelemetry.io/otel/sdk/export/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
+	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
+	processor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
+	selector "go.opentelemetry.io/otel/sdk/metric/selector/simple"
+	"go.opentelemetry.io/otel/sdk/resource"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+
+	"github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
+	"github.com/jaegertracing/jaeger-operator/pkg/tracing"
+)
+
+const meterName = "jaegertracing.io/jaeger"
+
+// Bootstrap configures the OpenTelemetry meter provider with the Prometheus exporter.
+func Bootstrap(ctx context.Context, namespace string, client client.Client) error {
+	tracer := otel.GetTracerProvider().Tracer(v1.BootstrapTracer)
+	ctx, span := tracer.Start(ctx, "bootstrap")
+	defer span.End()
+	tracing.SetInstanceID(ctx, namespace)
+
+	config := prometheus.Config{
+		Registry: metrics.Registry.(*prometheusclient.Registry),
+	}
+	c := controller.New(
+		processor.New(
+			selector.NewWithHistogramDistribution(
+				histogram.WithExplicitBoundaries(config.DefaultHistogramBoundaries),
+			),
+			export.CumulativeExportKindSelector(),
+			processor.WithMemory(true),
+		),
+		controller.WithResource(resource.NewWithAttributes([]attribute.KeyValue{}...)),
+	)
+	exporter, err := prometheus.NewExporter(config, c)
+	if err != nil {
+		return tracing.HandleError(err, span)
+	}
+
+	global.SetMeterProvider(exporter.MeterProvider())
+
+	// Create metrics
+	instancesObservedValue := newInstancesMetric(client)
+	err = instancesObservedValue.Setup(ctx)
+	return tracing.HandleError(err, span)
+}

pkg/metrics/instances.go

@@ -0,0 +1,160 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/metric/global"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	v1 "github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
+)
+
+const metricPrefix = "jaeger_operator_instances"
+const agentStrategiesMetric = "agent_strategies"
+const storageMetric = "storage_types"
+const strategiesMetric = "strategies"
+
+// This structure contains the labels associated with the instances and a counter of the number of instances
+type instancesView struct {
+	Name     string
+	Label    string
+	Count    map[string]int
+	Observer *metric.Int64ValueObserver
+	KeyFn    func(jaeger v1.Jaeger) string
+}
+
+func (i *instancesView) reset() {
+	for k := range i.Count {
+		i.Count[k] = 0
+	}
+}
+
+func (i *instancesView) Record(jaeger v1.Jaeger) {
+	i.Count[i.KeyFn(jaeger)]++
+}
+
+func (i *instancesView) Report(result metric.BatchObserverResult) {
+	for key, count := range i.Count {
+		result.Observe([]attribute.KeyValue{
+			attribute.String(i.Label, key),
+		}, i.Observer.Observation(int64(count)))
+	}
+}
+
+type instancesMetric struct {
+	client       client.Client
+	observations []instancesView
+}
+
+func instanceMetricName(name string) string {
+	return fmt.Sprintf("%s_%s", metricPrefix, name)
+}
+
+func newInstancesMetric(client client.Client) *instancesMetric {
+	return &instancesMetric{
+		client: client,
+	}
+}
+
+func newObservation(batch metric.BatchObserver, name, desc, label string, keyFn func(jaeger v1.Jaeger) string) (instancesView, error) {
+	observation := instancesView{
+		Name:  name,
+		Count: make(map[string]int),
+		KeyFn: keyFn,
+		Label: label,
+	}
+	obs, err := batch.NewInt64ValueObserver(instanceMetricName(name), metric.WithDescription(desc))
+	if err != nil {
+		return instancesView{}, err
+	}
+	observation.Observer = &obs
+	return observation, nil
+}
+
+func (i *instancesMetric) Setup(ctx context.Context) error {
+	tracer := otel.GetTracerProvider().Tracer(v1.BootstrapTracer)
+	ctx, span := tracer.Start(ctx, "setup-jaeger-instances-metrics")
+	defer span.End()
+	meter := global.Meter(meterName)
+	batch := meter.NewBatchObserver(i.callback)
+	obs, err := newObservation(batch,
+		agentStrategiesMetric,
+		"Number of instances per agent strategy",
+		"type",
+		func(jaeger v1.Jaeger) string {
+			return strings.ToLower(string(jaeger.Spec.Agent.Strategy))
+		})
+
+	if err != nil {
+		return err
+	}
+	i.observations = append(i.observations, obs)
+	obs, err = newObservation(batch, storageMetric,
+		"Number of instances per storage type",
+		"type",
+		func(jaeger v1.Jaeger) string {
+			return strings.ToLower(string(jaeger.Spec.Storage.Type))
+		})
+	if err != nil {
+		return err
+	}
+	i.observations = append(i.observations, obs)
+
+	obs, err = newObservation(batch, strategiesMetric,
+		"Number of instances per strategy type",
+		"type",
+		func(jaeger v1.Jaeger) string {
+			return strings.ToLower(string(jaeger.Spec.Strategy))
+		})
+	if err != nil {
+		return err
+	}
+	i.observations = append(i.observations, obs)
+	return nil
+}
+
+func isInstanceNormalized(jaeger v1.Jaeger) bool {
+	return !(jaeger.Spec.Storage.Type == "" || jaeger.Spec.Strategy == "")
+}
+
+func normalizeAgentStrategy(jaeger *v1.Jaeger) {
+	if jaeger.Spec.Agent.Strategy == "" {
+		jaeger.Spec.Agent.Strategy = "Sidecar"
+	}
+}
+
+func (i *instancesMetric) reset() {
+	for _, o := range i.observations {
+		o.reset()
+	}
+}
+
+func (i *instancesMetric) report(result metric.BatchObserverResult) {
+	for _, o := range i.observations {
+		o.Report(result)
+	}
+}
+
+func (i *instancesMetric) callback(ctx context.Context, result metric.BatchObserverResult) {
+	instances := &v1.JaegerList{}
+	if err := i.client.List(ctx, instances); err == nil {
+		i.reset()
+		for _, jaeger := range instances.Items {
+			// Is this instance is already normalized by the reconciliation process
+			// count it on the metrics, otherwise not.
+			if isInstanceNormalized(jaeger) {
+				// Normalization doesn't normalize agent mode. so we need to do it here.
+				normalizeAgentStrategy(&jaeger)
+				for _, o := range i.observations {
+					o.Record(jaeger)
+				}
+			}
+		}
+		i.report(result)
+	}
+}