feat(observability): allow custom latency distribution buckets

WIP

pkg/apis/configuration/v1alpha1/types.go

@@ -222,6 +222,9 @@ type MetricHTTP struct {
  // If false (the default), metrics for the HTTP server are collected with increased cardinality.
  // +optional
  IncreasedCardinality *bool `json:"increasedCardinality,omitempty"`
+ // Latency distribution buckets. If not set, the default buckets are used.
+ // +optional
+ LatencyDistributionBuckets []int `json:"latencyDistributionBuckets,omitempty"`
 }
 
 // MetricsRule defines configuration options for a metric.

pkg/config/configuration.go

@@ -269,10 +269,28 @@ func (m MetricSpec) GetHTTPIncreasedCardinality(log logger.Logger) bool {
  return *m.HTTP.IncreasedCardinality
 }
 
+// GetLatencyDistributionBuckets returns a []float64 to be used true for distribution latency buckets
+func (m MetricSpec) GetLatencyDistributionBuckets(log logger.Logger) []float64 {
+ var defaultLatencyDistribution = []float64{1, 2, 3, 4, 5, 6, 8, 10, 13, 16, 20, 25, 30, 40, 50, 65, 80, 100, 130, 160, 200, 250, 300, 400, 500, 650, 800, 1_000, 2_000, 5_000, 10_000, 20_000, 50_000, 100_000}
+ if m.HTTP == nil || m.HTTP.LatencyDistributionBuckets == nil {
+ // The default is defaultLatencyDistribution
+ log.Infof("Using default latency distribution buckets: %v", defaultLatencyDistribution)
+ return defaultLatencyDistribution
+ }
+ log.Infof("Using custom latency distribution buckets: %v", *m.HTTP.LatencyDistributionBuckets)
+ floats := make([]float64, len(*m.HTTP.LatencyDistributionBuckets))
+ for i, v := range *m.HTTP.LatencyDistributionBuckets {
+ floats[i] = float64(v)
+ }
+ return floats
+}
+
 // MetricHTTP defines configuration for metrics for the HTTP server
 type MetricHTTP struct {
  // If false (the default), metrics for the HTTP server are collected with increased cardinality.
  IncreasedCardinality *bool `json:"increasedCardinality,omitempty" yaml:"increasedCardinality,omitempty"`
+ // Latency distribution buckets. If not set, the default buckets are used.
+ LatencyDistributionBuckets *[]int `json:"latencyDistributionBuckets,omitempty"`
 }
 
 // MetricsRu le defines configuration options for a metric.

pkg/diagnostics/component_monitoring.go

@@ -161,30 +161,30 @@ func newComponentMetrics() *componentMetrics {
 }
 
 // Init registers the component metrics views.
-func (c *componentMetrics) Init(appID, namespace string) error {
+func (c *componentMetrics) Init(appID, namespace string, latencyDistribution *view.Aggregation) error {
  c.appID = appID
  c.enabled = true
  c.namespace = namespace
 
  return view.Register(
- diagUtils.NewMeasureView(c.pubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.pubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.pubsubIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, view.Count()),
- diagUtils.NewMeasureView(c.bulkPubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.bulkPubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.bulkPubsubIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, view.Count()),
  diagUtils.NewMeasureView(c.bulkPubsubEventIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, view.Count()),
- diagUtils.NewMeasureView(c.pubsubEgressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, successKey, topicKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.pubsubEgressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, successKey, topicKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.pubsubEgressCount, []tag.Key{appIDKey, componentKey, namespaceKey, successKey, topicKey}, view.Count()),
- diagUtils.NewMeasureView(c.inputBindingLatency, []tag.Key{appIDKey, componentKey, namespaceKey, successKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.inputBindingLatency, []tag.Key{appIDKey, componentKey, namespaceKey, successKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.inputBindingCount, []tag.Key{appIDKey, componentKey, namespaceKey, successKey}, view.Count()),
- diagUtils.NewMeasureView(c.outputBindingLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.outputBindingLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.outputBindingCount, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, view.Count()),
- diagUtils.NewMeasureView(c.stateLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.stateLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.stateCount, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, view.Count()),
- diagUtils.NewMeasureView(c.configurationLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.configurationLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.configurationCount, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, view.Count()),
- diagUtils.NewMeasureView(c.secretLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.secretLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.secretCount, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, view.Count()),
- diagUtils.NewMeasureView(c.cryptoLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(c.cryptoLatency, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, latencyDistribution),
  diagUtils.NewMeasureView(c.cryptoCount, []tag.Key{appIDKey, componentKey, namespaceKey, operationKey, successKey}, view.Count()),
  )
 }

pkg/diagnostics/component_monitoring_test.go

@@ -15,7 +15,8 @@ const (
 
 func componentsMetrics() *componentMetrics {
  c := newComponentMetrics()
- c.Init("test", "default")
+ latencyDistribution := view.Distribution(5, 50, 500, 5_000)
+ c.Init("test", "default", latencyDistribution)
 
  return c
 }

pkg/diagnostics/grpc_monitoring.go

@@ -109,20 +109,20 @@ func newGRPCMetrics() *grpcMetrics {
  }
 }
 
-func (g *grpcMetrics) Init(appID string) error {
+func (g *grpcMetrics) Init(appID string, latencyDistribution *view.Aggregation) error {
  g.appID = appID
  g.enabled = true
 
  return view.Register(
  diagUtils.NewMeasureView(g.serverReceivedBytes, []tag.Key{appIDKey, KeyServerMethod}, defaultSizeDistribution),
  diagUtils.NewMeasureView(g.serverSentBytes, []tag.Key{appIDKey, KeyServerMethod}, defaultSizeDistribution),
- diagUtils.NewMeasureView(g.serverLatency, []tag.Key{appIDKey, KeyServerMethod, KeyServerStatus}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(g.serverLatency, []tag.Key{appIDKey, KeyServerMethod, KeyServerStatus}, latencyDistribution),
  diagUtils.NewMeasureView(g.serverCompletedRpcs, []tag.Key{appIDKey, KeyServerMethod, KeyServerStatus}, view.Count()),
  diagUtils.NewMeasureView(g.clientSentBytes, []tag.Key{appIDKey, KeyClientMethod}, defaultSizeDistribution),
  diagUtils.NewMeasureView(g.clientReceivedBytes, []tag.Key{appIDKey, KeyClientMethod}, defaultSizeDistribution),
- diagUtils.NewMeasureView(g.clientRoundtripLatency, []tag.Key{appIDKey, KeyClientMethod, KeyClientStatus}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(g.clientRoundtripLatency, []tag.Key{appIDKey, KeyClientMethod, KeyClientStatus}, latencyDistribution),
  diagUtils.NewMeasureView(g.clientCompletedRpcs, []tag.Key{appIDKey, KeyClientMethod, KeyClientStatus}, view.Count()),
- diagUtils.NewMeasureView(g.healthProbeRoundripLatency, []tag.Key{appIDKey, KeyClientStatus}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(g.healthProbeRoundripLatency, []tag.Key{appIDKey, KeyClientStatus}, latencyDistribution),
  diagUtils.NewMeasureView(g.healthProbeCompletedCount, []tag.Key{appIDKey, KeyClientStatus}, view.Count()),
  )
 }

pkg/diagnostics/grpc_monitoring_test.go

@@ -61,9 +61,11 @@ func (f *fakeProxyStream) RecvMsg(m interface{}) error {
 }
 
 func TestStreamingServerInterceptor(t *testing.T) {
+ latencyDistribution := view.Distribution(5, 50, 500, 5_000)
+
  t.Run("not a proxy request, do not run pipeline", func(t *testing.T) {
  m := newGRPCMetrics()
- m.Init("test")
+ m.Init("test", latencyDistribution)
 
  i := m.StreamingServerInterceptor()
  s := &fakeProxyStream{}
@@ -85,7 +87,7 @@ func TestStreamingServerInterceptor(t *testing.T) {
 
  t.Run("proxy request, run pipeline", func(t *testing.T) {
  m := newGRPCMetrics()
- m.Init("test")
+ m.Init("test", latencyDistribution)
 
  i := m.StreamingServerInterceptor()
  s := &fakeProxyStream{
@@ -115,9 +117,10 @@ func TestStreamingServerInterceptor(t *testing.T) {
 }
 
 func TestStreamingClientInterceptor(t *testing.T) {
+ latencyDistribution := view.Distribution(5, 50, 500, 5_000)
  t.Run("not a proxy request, do not run pipeline", func(t *testing.T) {
  m := newGRPCMetrics()
- m.Init("test")
+ m.Init("test", latencyDistribution)
 
  i := m.StreamingClientInterceptor()
  s := &fakeProxyStream{}
@@ -139,7 +142,7 @@ func TestStreamingClientInterceptor(t *testing.T) {
 
  t.Run("proxy request, run pipeline", func(t *testing.T) {
  m := newGRPCMetrics()
- m.Init("test")
+ m.Init("test", latencyDistribution)
 
  i := m.StreamingClientInterceptor()
  s := &fakeProxyStream{

pkg/diagnostics/http_monitoring.go

@@ -40,8 +40,7 @@ var (
 
 var (
  // <<10 -> KBs; <<20 -> MBs; <<30 -> GBs
- defaultSizeDistribution = view.Distribution(1<<10, 2<<10, 4<<10, 16<<10, 64<<10, 256<<10, 1<<20, 4<<20, 16<<20, 64<<20, 256<<20, 1<<30, 4<<30)
- defaultLatencyDistribution = view.Distribution(1, 2, 3, 4, 5, 6, 8, 10, 13, 16, 20, 25, 30, 40, 50, 65, 80, 100, 130, 160, 200, 250, 300, 400, 500, 650, 800, 1_000, 2_000, 5_000, 10_000, 20_000, 50_000, 100_000)
+ defaultSizeDistribution = view.Distribution(1<<10, 2<<10, 4<<10, 16<<10, 64<<10, 256<<10, 1<<20, 4<<20, 16<<20, 64<<20, 256<<20, 1<<30, 4<<30)
 )
 
 type httpMetrics struct {
@@ -218,7 +217,7 @@ func (h *httpMetrics) AppHealthProbeCompleted(ctx context.Context, status string
  h.healthProbeRoundripLatency.M(elapsed))
 }
 
-func (h *httpMetrics) Init(appID string, legacy bool) error {
+func (h *httpMetrics) Init(appID string, legacy bool, latencyDistribution *view.Aggregation) error {
  h.appID = appID
  h.enabled = true
  h.legacy = legacy
@@ -237,13 +236,13 @@ func (h *httpMetrics) Init(appID string, legacy bool) error {
  return view.Register(
  diagUtils.NewMeasureView(h.serverRequestBytes, tags, defaultSizeDistribution),
  diagUtils.NewMeasureView(h.serverResponseBytes, tags, defaultSizeDistribution),
- diagUtils.NewMeasureView(h.serverLatency, serverTags, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(h.serverLatency, serverTags, latencyDistribution),
  diagUtils.NewMeasureView(h.serverRequestCount, serverTags, view.Count()),
  diagUtils.NewMeasureView(h.clientSentBytes, clientTags, defaultSizeDistribution),
  diagUtils.NewMeasureView(h.clientReceivedBytes, tags, defaultSizeDistribution),
- diagUtils.NewMeasureView(h.clientRoundtripLatency, clientTags, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(h.clientRoundtripLatency, clientTags, latencyDistribution),
  diagUtils.NewMeasureView(h.clientCompletedCount, clientTags, view.Count()),
- diagUtils.NewMeasureView(h.healthProbeRoundripLatency, []tag.Key{appIDKey, httpStatusCodeKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(h.healthProbeRoundripLatency, []tag.Key{appIDKey, httpStatusCodeKey}, latencyDistribution),
  diagUtils.NewMeasureView(h.healthProbeCompletedCount, []tag.Key{appIDKey, httpStatusCodeKey}, view.Count()),
  )
 }

pkg/diagnostics/http_monitoring_test.go

@@ -21,7 +21,8 @@ func TestHTTPMiddleware(t *testing.T) {
 
  // create test httpMetrics
  testHTTP := newHTTPMetrics()
- testHTTP.Init("fakeID", false)
+ latencyDistribution := view.Distribution(5, 50, 500, 5_000)
+ testHTTP.Init("fakeID", false, latencyDistribution)
 
  handler := testHTTP.HTTPMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  time.Sleep(100 * time.Millisecond)
@@ -67,8 +68,9 @@ func TestHTTPMiddlewareWhenMetricsDisabled(t *testing.T) {
  // create test httpMetrics
  testHTTP := newHTTPMetrics()
  testHTTP.enabled = false
+ latencyDistribution := view.Distribution(5, 50, 500, 5_000)
 
- testHTTP.Init("fakeID", false)
+ testHTTP.Init("fakeID", false, latencyDistribution)
  v := view.Find("http/server/request_count")
  views := []*view.View{v}
  view.Unregister(views...)

pkg/diagnostics/metrics.go

@@ -45,28 +45,31 @@ var (
 )
 
 // InitMetrics initializes metrics.
-func InitMetrics(appID, namespace string, rules []config.MetricsRule, legacyMetricsHTTPMetrics bool) error {
- if err := DefaultMonitoring.Init(appID); err != nil {
+func InitMetrics(appID, namespace string, rules []config.MetricsRule, legacyMetricsHTTPMetrics bool, latencyDistributionBuckets []float64) error {
+ // Initialize metrics
+ latencyDistribution := view.Distribution(latencyDistributionBuckets...)
+
+ if err := DefaultMonitoring.Init(appID, latencyDistribution); err != nil {
  return err
  }
 
- if err := DefaultGRPCMonitoring.Init(appID); err != nil {
+ if err := DefaultGRPCMonitoring.Init(appID, latencyDistribution); err != nil {
  return err
  }
 
- if err := DefaultHTTPMonitoring.Init(appID, legacyMetricsHTTPMetrics); err != nil {
+ if err := DefaultHTTPMonitoring.Init(appID, legacyMetricsHTTPMetrics, latencyDistribution); err != nil {
  return err
  }
 
- if err := DefaultComponentMonitoring.Init(appID, namespace); err != nil {
+ if err := DefaultComponentMonitoring.Init(appID, namespace, latencyDistribution); err != nil {
  return err
  }
 
  if err := DefaultResiliencyMonitoring.Init(appID); err != nil {
  return err
  }
 
- if err := DefaultWorkflowMonitoring.Init(appID, namespace); err != nil {
+ if err := DefaultWorkflowMonitoring.Init(appID, namespace, latencyDistribution); err != nil {
  return err
  }
 

pkg/diagnostics/metrics_regex_test.go

@@ -17,6 +17,8 @@ func TestRegexRulesSingle(t *testing.T) {
  const statName = "test_stat_regex"
  methodKey := tag.MustNewKey("method")
  testStat := stats.Int64(statName, "Stat used in unit test", stats.UnitDimensionless)
+ latencyDistributionBuckets := []float64{5, 50, 500, 5_000}
+ latencyDistribution := view.Distribution(latencyDistributionBuckets...)
 
  InitMetrics("testAppId2", "", []config.MetricsRule{
  {
@@ -31,7 +33,7 @@ func TestRegexRulesSingle(t *testing.T) {
  },
  },
  },
- }, false)
+ }, false, latencyDistributionBuckets)
 
  t.Run("single regex rule applied", func(t *testing.T) {
  view.Register(
@@ -62,7 +64,7 @@ func TestRegexRulesSingle(t *testing.T) {
  })
 
  s := newGRPCMetrics()
- s.Init("test")
+ s.Init("test", latencyDistribution)
 
  stats.RecordWithTags(context.Background(),
  diagUtils.WithTags(testStat.Name(), methodKey, "/siths/123"),
@@ -85,7 +87,7 @@ func TestRegexRulesSingle(t *testing.T) {
  })
 
  s := newGRPCMetrics()
- s.Init("test")
+ s.Init("test", latencyDistribution)
 
  stats.RecordWithTags(context.Background(),
  diagUtils.WithTags(testStat.Name(), methodKey, "/orders/123"),

pkg/diagnostics/service_monitoring.go

@@ -209,7 +209,7 @@ func newServiceMetrics() *serviceMetrics {
 }
 
 // Init initialize metrics views for metrics.
-func (s *serviceMetrics) Init(appID string) error {
+func (s *serviceMetrics) Init(appID string, latencyDistribution *view.Aggregation) error {
  s.appID = appID
  s.enabled = true
  return view.Register(
@@ -243,7 +243,7 @@ func (s *serviceMetrics) Init(appID string) error {
  diagUtils.NewMeasureView(s.serviceInvocationRequestReceivedTotal, []tag.Key{appIDKey, sourceAppIDKey}, view.Count()),
  diagUtils.NewMeasureView(s.serviceInvocationResponseSentTotal, []tag.Key{appIDKey, destinationAppIDKey, statusKey}, view.Count()),
  diagUtils.NewMeasureView(s.serviceInvocationResponseReceivedTotal, []tag.Key{appIDKey, sourceAppIDKey, statusKey, typeKey}, view.Count()),
- diagUtils.NewMeasureView(s.serviceInvocationResponseReceivedLatency, []tag.Key{appIDKey, sourceAppIDKey, statusKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(s.serviceInvocationResponseReceivedLatency, []tag.Key{appIDKey, sourceAppIDKey, statusKey}, latencyDistribution),
  )
 }
 

pkg/diagnostics/service_monitoring_test.go

@@ -10,7 +10,8 @@ import (
 
 func servicesMetrics() *serviceMetrics {
  s := newServiceMetrics()
- s.Init("testAppId")
+ latencyDistribution := view.Distribution(5, 50, 500, 5_000)
+ s.Init("testAppId", latencyDistribution)
 
  return s
 }

pkg/diagnostics/workflow_monitoring.go

@@ -99,19 +99,19 @@ func (w *workflowMetrics) IsEnabled() bool {
 }
 
 // Init registers the workflow metrics views.
-func (w *workflowMetrics) Init(appID, namespace string) error {
+func (w *workflowMetrics) Init(appID, namespace string, latencyDistribution *view.Aggregation) error {
  w.appID = appID
  w.enabled = true
  w.namespace = namespace
 
  return view.Register(
  diagUtils.NewMeasureView(w.workflowOperationCount, []tag.Key{appIDKey, namespaceKey, operationKey, statusKey}, view.Count()),
- diagUtils.NewMeasureView(w.workflowOperationLatency, []tag.Key{appIDKey, namespaceKey, operationKey, statusKey}, defaultLatencyDistribution),
+ diagUtils.NewMeasureView(w.workflowOperationLatency, []tag.Key{appIDKey, namespaceKey, operationKey, statusKey}, latencyDistribution),
  diagUtils.NewMeasureView(w.workflowExecutionCount, []tag.Key{appIDKey, namespaceKey, workflowNameKey, statusKey}, view.Count()),
  diagUtils.NewMeasureView(w.activityExecutionCount, []tag.Key{appIDKey, namespaceKey, activityNameKey, statusKey}, view.Count()),
- diagUtils.NewMeasureView(w.activityExecutionLatency, []tag.Key{appIDKey, namespaceKey, activityNameKey, statusKey}, defaultLatencyDistribution),
- diagUtils.NewMeasureView(w.workflowExecutionLatency, []tag.Key{appIDKey, namespaceKey, workflowNameKey, statusKey}, defaultLatencyDistribution),
- diagUtils.NewMeasureView(w.workflowSchedulingLatency, []tag.Key{appIDKey, namespaceKey, workflowNameKey}, defaultLatencyDistribution))
+ diagUtils.NewMeasureView(w.activityExecutionLatency, []tag.Key{appIDKey, namespaceKey, activityNameKey, statusKey}, latencyDistribution),
+ diagUtils.NewMeasureView(w.workflowExecutionLatency, []tag.Key{appIDKey, namespaceKey, workflowNameKey, statusKey}, latencyDistribution),
+ diagUtils.NewMeasureView(w.workflowSchedulingLatency, []tag.Key{appIDKey, namespaceKey, workflowNameKey}, latencyDistribution))
 }
 
 // WorkflowOperationEvent records total number of Successful/Failed workflow Operations requests. It also records latency for those requests.