WIP: new metrics emitter and histogram strategy

Makefile

@@ -64,7 +64,7 @@ $(BUILD)/goversion-lint:
 $(BUILD)/fmt: $(BUILD)/codegen # formatting must occur only after all other go-file-modifications are done
 # $(BUILD)/copyright 
 # $(BUILD)/copyright: $(BUILD)/codegen # must add copyright to generated code, sometimes needs re-formatting
-$(BUILD)/codegen: $(BUILD)/thrift $(BUILD)/protoc
+$(BUILD)/codegen: $(BUILD)/thrift $(BUILD)/protoc $(BUILD)/metrics
 $(BUILD)/thrift: $(BUILD)/go_mod_check
 $(BUILD)/protoc: $(BUILD)/go_mod_check
 $(BUILD)/go_mod_check:
@@ -211,6 +211,12 @@ $(BIN)/protoc-gen-gogofast: go.mod go.work | $(BIN)
 $(BIN)/protoc-gen-yarpc-go: go.mod go.work | $(BIN)
 	$(call go_mod_build_tool,go.uber.org/yarpc/encoding/protobuf/protoc-gen-yarpc-go)
 
+$(BIN)/metricsgen: internal/tools/go.mod go.work $(wildcard internal/tools/metricsgen/*) | $(BIN)
+	$(call go_build_tool,./metricsgen)
+
+$(BIN)/metricslint: internal/tools/go.mod go.work $(wildcard internal/tools/metricslint/* internal/tools/metricslint/cmd/*) | $(BIN)
+	$(call go_build_tool,./metricslint/cmd,metricslint)
+
 $(BUILD)/go_mod_check: go.mod internal/tools/go.mod go.work
 	$Q # generated == used is occasionally important for gomock / mock libs in general.  this is not a definite problem if violated though.
 	$Q ./scripts/check-gomod-version.sh github.com/golang/mock/gomock $(if $(verbose),-v)
@@ -352,6 +358,10 @@ $(BUILD)/protoc: $(PROTO_FILES) $(STABLE_BIN)/$(PROTOC_VERSION_BIN) $(BIN)/proto
 	   fi
 	$Q touch $@
 
+$(BUILD)/metrics: $(ALL_SRC) $(BIN)/metricsgen
+	$Q $(BIN_PATH) go generate -run=metricsgen ./...
+	$Q touch $@
+
 # ====================================
 # Rule-breaking targets intended ONLY for special cases with no good alternatives.
 # ====================================
@@ -404,6 +414,11 @@ $(BUILD)/code-lint: $(LINT_SRC) $(BIN)/revive | $(BUILD)
 		fi
 	$Q touch $@
 
+$(BUILD)/metrics-lint: $(ALL_SRC) $(BIN)/metricslint | $(BUILD)
+	$Q echo "linting metrics definitions..."
+	$Q $(BIN_PATH) $(BIN)/metricslint -skip cadence_requests_per_tl,2 -skip cache_hit,2 -skip cache_full,2 -skip cache_miss,2 -skip cross_cluster_fetch_errors,2 ./...
+	$Q touch $@
+
 $(BUILD)/goversion-lint: go.work Dockerfile docker/github_actions/Dockerfile${DOCKERFILE_SUFFIX}
 	$Q echo "checking go version..."
 	$Q # intentionally using go.work toolchain, as GOTOOLCHAIN is user-overridable
@@ -458,7 +473,7 @@ endef
 # useful to actually re-run to get output again.
 # reuse the intermediates for simplicity and consistency.
 lint: ## (Re)run the linter
-	$(call remake,proto-lint gomod-lint code-lint goversion-lint)
+	$(call remake,proto-lint gomod-lint code-lint goversion-lint metrics-lint)
 
 # intentionally not re-making, it's a bit slow and it's clear when it's unnecessary
 fmt: $(BUILD)/fmt ## Run `gofmt` / organize imports / etc
@@ -544,14 +559,20 @@ bins: $(BINS) ## Build all binaries, and any fast codegen needed (does not refre
 
 tools: $(TOOLS)
 
-go-generate: $(BIN)/mockgen $(BIN)/enumer $(BIN)/mockery  $(BIN)/gowrap ## Run `go generate` to regen mocks, enums, etc
+go-generate: $(BIN)/mockgen $(BIN)/enumer $(BIN)/mockery  $(BIN)/gowrap $(BIN)/metricsgen ## Run `go generate` to regen mocks, enums, etc
 	$Q echo "running go generate ./..., this takes a minute or more..."
 	$Q # add our bins to PATH so `go generate` can find them
 	$Q $(BIN_PATH) go generate $(if $(verbose),-v) ./...
+	$Q touch $(BUILD)/metrics # whole-service go-generate also regenerates metrics
 	$Q $(MAKE) --no-print-directory fmt
 # 	$Q echo "updating copyright headers"
 # 	$Q $(MAKE) --no-print-directory copyright
 
+metrics: $(BIN)/metricsgen ## metrics-only code regen, much faster than go-generate
+	$Q echo "re-generating metrics structs..."
+	$Q $(MAKE) $(BUILD)/metrics
+	$Q $(MAKE) fmt # clean up imports
+
 release: ## Re-generate generated code and run tests
 	$(MAKE) --no-print-directory go-generate
 	$(MAKE) --no-print-directory test
@@ -577,7 +598,7 @@ tidy: ## `go mod tidy` all packages
 clean: ## Clean build products and SQLite database
 	rm -f $(BINS)
 	rm -Rf $(BUILD)
-	rm *.db
+	rm -f *.db
 	$(if \
 		$(wildcard $(STABLE_BIN)/*), \
 		$(warning usually-stable build tools still exist, delete the $(STABLE_BIN) folder to rebuild them),)

cmd/server/cadence/fx.go

@@ -40,6 +40,7 @@ import (
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
 	"github.com/uber/cadence/common/metrics/metricsfx"
+	"github.com/uber/cadence/common/metrics/structured"
 	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
 	"github.com/uber/cadence/common/rpc/rpcfx"
 	"github.com/uber/cadence/common/service"
@@ -101,6 +102,7 @@ type AppParams struct {
 	DynamicConfig dynamicconfig.Client
 	Scope         tally.Scope
 	MetricsClient metrics.Client
+	Emitter       structured.Emitter
 }
 
 // NewApp created a new Application from pre initalized config and logger.
@@ -112,6 +114,7 @@ func NewApp(params AppParams) *App {
 		dynamicConfig: params.DynamicConfig,
 		scope:         params.Scope,
 		metricsClient: params.MetricsClient,
+		emitter:       params.Emitter,
 	}
 
 	params.LifeCycle.Append(fx.StartHook(app.verifySchema))
@@ -128,13 +131,14 @@ type App struct {
 	dynamicConfig dynamicconfig.Client
 	scope         tally.Scope
 	metricsClient metrics.Client
+	emitter       structured.Emitter
 
 	daemon  common.Daemon
 	service string
 }
 
 func (a *App) Start(_ context.Context) error {
-	a.daemon = newServer(a.service, a.cfg, a.logger, a.dynamicConfig, a.scope, a.metricsClient)
+	a.daemon = newServer(a.service, a.cfg, a.logger, a.dynamicConfig, a.scope, a.metricsClient, a.emitter)
 	a.daemon.Start()
 	return nil
 }

cmd/server/cadence/server.go

@@ -54,6 +54,7 @@ import (
 	"github.com/uber/cadence/common/membership"
 	"github.com/uber/cadence/common/messaging/kafka"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	"github.com/uber/cadence/common/peerprovider/ringpopprovider"
 	pnt "github.com/uber/cadence/common/pinot"
 	"github.com/uber/cadence/common/resource"
@@ -79,12 +80,13 @@ type (
 		dynamicCfgClient dynamicconfig.Client
 		scope            tally.Scope
 		metricsClient    metrics.Client
+		emitter          structured.Emitter
 	}
 )
 
 // newServer returns a new instance of a daemon
 // that represents a cadence service
-func newServer(service string, cfg config.Config, logger log.Logger, dynamicCfgClient dynamicconfig.Client, scope tally.Scope, metricsClient metrics.Client) common.Daemon {
+func newServer(service string, cfg config.Config, logger log.Logger, dynamicCfgClient dynamicconfig.Client, scope tally.Scope, metricsClient metrics.Client, emitter structured.Emitter) common.Daemon {
 	return &server{
 		cfg:              cfg,
 		name:             service,
@@ -93,6 +95,7 @@ func newServer(service string, cfg config.Config, logger log.Logger, dynamicCfgC
 		dynamicCfgClient: dynamicCfgClient,
 		scope:            scope,
 		metricsClient:    metricsClient,
+		emitter:          emitter,
 	}
 }
 
@@ -142,6 +145,7 @@ func (s *server) startService() common.Daemon {
 
 	params.MetricScope = s.scope
 	params.MetricsClient = s.metricsClient
+	params.Emitter = s.emitter
 
 	rpcParams, err := rpc.NewParams(params.Name, &s.cfg, dc, params.Logger, params.MetricsClient)
 	if err != nil {

cmd/server/cadence/server_test.go

@@ -44,6 +44,7 @@ import (
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/log/testlogger"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	pt "github.com/uber/cadence/common/persistence/persistence-tests"
 	"github.com/uber/cadence/common/persistence/sql/sqlplugin/sqlite"
 	"github.com/uber/cadence/common/resource"
@@ -110,7 +111,15 @@ func (s *ServerSuite) TestServerStartup() {
 		})
 
 	for _, svc := range services {
-		server := newServer(svc, cfg, logger, dynamicconfig.NewNopClient(), tally.NoopScope, metrics.NewNoopMetricsClient())
+		server := newServer(
+			svc,
+			cfg,
+			logger,
+			dynamicconfig.NewNopClient(),
+			tally.NoopScope,
+			metrics.NewNoopMetricsClient(),
+			structured.NewTestEmitter(s.T(), nil),
+		)
 		daemons = append(daemons, server)
 		server.Start()
 	}

common/metrics/defs.go

@@ -53,6 +53,10 @@ type (
 	ServiceIdx int
 )
 
+func (s scopeDefinition) GetOperationString() string {
+	return s.operation
+}
+
 // MetricTypes which are supported
 const (
 	Counter MetricType = iota
@@ -1068,7 +1072,7 @@ const (
 // -- Operation scopes for History service --
 const (
 	// HistoryStartWorkflowExecutionScope tracks StartWorkflowExecution API calls received by service
-	HistoryStartWorkflowExecutionScope = iota + NumCommonScopes
+	HistoryStartWorkflowExecutionScope = iota + NumFrontendScopes
 	// HistoryRecordActivityTaskHeartbeatScope tracks RecordActivityTaskHeartbeat API calls received by service
 	HistoryRecordActivityTaskHeartbeatScope
 	// HistoryRespondDecisionTaskCompletedScope tracks RespondDecisionTaskCompleted API calls received by service
@@ -1356,7 +1360,7 @@ const (
 // -- Operation scopes for Matching service --
 const (
 	// PollForDecisionTaskScope tracks PollForDecisionTask API calls received by service
-	MatchingPollForDecisionTaskScope = iota + NumCommonScopes
+	MatchingPollForDecisionTaskScope = iota + NumHistoryScopes
 	// PollForActivityTaskScope tracks PollForActivityTask API calls received by service
 	MatchingPollForActivityTaskScope
 	// MatchingAddActivityTaskScope tracks AddActivityTask API calls received by service
@@ -1392,7 +1396,7 @@ const (
 // -- Operation scopes for Worker service --
 const (
 	// ReplicationScope is the scope used by all metric emitted by replicator
-	ReplicatorScope = iota + NumCommonScopes
+	ReplicatorScope = iota + NumMatchingScopes
 	// DomainReplicationTaskScope is the scope used by domain task replication processing
 	DomainReplicationTaskScope
 	// ESProcessorScope is scope used by all metric emitted by esProcessor
@@ -1440,7 +1444,7 @@ const (
 // -- Operation scopes for ShardDistributor service --
 const (
 	// ShardDistributorGetShardOwnerScope tracks GetShardOwner API calls received by service
-	ShardDistributorGetShardOwnerScope = iota + NumCommonScopes
+	ShardDistributorGetShardOwnerScope = iota + NumWorkerScopes
 	ShardDistributorHeartbeatScope
 	ShardDistributorAssignLoopScope
 
@@ -2700,12 +2704,13 @@ const (
 	VirtualQueueCountGauge
 	VirtualQueuePausedGauge
 	VirtualQueueRunningGauge
+
 	NumHistoryMetrics
 )
 
 // Matching metrics enum
 const (
-	PollSuccessPerTaskListCounter = iota + NumCommonMetrics
+	PollSuccessPerTaskListCounter = iota + NumHistoryMetrics
 	PollTimeoutPerTaskListCounter
 	PollSuccessWithSyncPerTaskListCounter
 	LeaseRequestPerTaskListCounter
@@ -2783,12 +2788,13 @@ const (
 	IsolationGroupUpscale
 	IsolationGroupDownscale
 	PartitionDrained
+
 	NumMatchingMetrics
 )
 
 // Worker metrics enum
 const (
-	ReplicatorMessages = iota + NumCommonMetrics
+	ReplicatorMessages = iota + NumMatchingMetrics
 	ReplicatorFailures
 	ReplicatorMessagesDropped
 	ReplicatorLatency
@@ -2872,12 +2878,13 @@ const (
 	DiagnosticsWorkflowStartedCount
 	DiagnosticsWorkflowSuccess
 	DiagnosticsWorkflowExecutionLatency
+
 	NumWorkerMetrics
 )
 
 // ShardDistributor metrics enum
 const (
-	ShardDistributorRequests = iota + NumCommonMetrics
+	ShardDistributorRequests = iota + NumWorkerMetrics
 	ShardDistributorFailures
 	ShardDistributorLatency
 	ShardDistributorErrContextTimeoutCounter

common/metrics/defs_test.go

@@ -129,6 +129,39 @@ func TestMetricDefs(t *testing.T) {
 	}
 }
 
+// "index -> operation" must be unique for structured.DynamicOperationTags' int lookup to work consistently.
+// Duplicate indexes with the same operation name are technically fine, but there doesn't seem to be any benefit in allowing it,
+// and it trivially ensures that all indexes have only one operation name.
+func TestOperationIndexesAreUnique(t *testing.T) {
+	seen := make(map[int]bool)
+	for serviceIdx, serviceOps := range ScopeDefs {
+		for idx := range serviceOps {
+			if seen[idx] {
+				t.Error("duplicate operation index:", idx, "with name:", serviceOps[idx].operation, "in service:", serviceIdx)
+			}
+			seen[idx] = true
+		}
+	}
+}
+
+func TestMetricsAreUnique(t *testing.T) {
+	// Duplicate indexes is arguably fine, but there doesn't seem to be any benefit in allowing it.
+	//
+	// Duplicate names are also linted, but they're done via an analyzer (metricslint) instead, to
+	// allow checking across multiple formats.
+	t.Run("indexes", func(t *testing.T) {
+		seen := make(map[int]bool)
+		for _, serviceMetrics := range MetricDefs {
+			for idx := range serviceMetrics {
+				if seen[idx] {
+					t.Error("duplicate metric index:", idx, "with name:", serviceMetrics[idx].metricName)
+				}
+				seen[idx] = true
+			}
+		}
+	})
+}
+
 func TestExponentialDurationBuckets(t *testing.T) {
 	factor := math.Pow(2, 0.25)
 	assert.Equal(t, 80, len(ExponentialDurationBuckets))

common/metrics/metricsfx/metricsfx.go

@@ -29,12 +29,15 @@ import (
 	"github.com/uber/cadence/common/config"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	"github.com/uber/cadence/common/service"
 )
 
 // Module provides metrics client for fx application.
 var Module = fx.Module("metricsfx",
-	fx.Provide(buildClient))
+	fx.Provide(buildClient),
+	structured.Module,
+)
 
 // ModuleForExternalScope provides metrics client for fx application when tally.Scope is created outside.
 var ModuleForExternalScope = fx.Module("metricsfx",