use the new metrics thing

Author: Groxx

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/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",

common/resource/params.go

@@ -42,6 +42,7 @@ import (
 	"github.com/uber/cadence/common/membership"
 	"github.com/uber/cadence/common/messaging"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	persistenceClient "github.com/uber/cadence/common/persistence/client"
 	"github.com/uber/cadence/common/pinot"
 	"github.com/uber/cadence/common/rpc"
@@ -68,6 +69,7 @@ type (
 		ClusterMetadata    cluster.Metadata
 		ReplicatorConfig   config.Replicator
 		MetricsClient      metrics.Client
+		Emitter            structured.Emitter
 		MessagingClient    messaging.Client
 		BlobstoreClient    blobstore.Client
 		ESClient           es.GenericClient

common/resource/resource_impl.go

@@ -57,6 +57,7 @@ import (
 	"github.com/uber/cadence/common/membership"
 	"github.com/uber/cadence/common/messaging"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	"github.com/uber/cadence/common/persistence"
 	persistenceClient "github.com/uber/cadence/common/persistence/client"
 	qrpc "github.com/uber/cadence/common/quotas/global/rpc"
@@ -98,6 +99,7 @@ type Impl struct {
 	timeSource              clock.TimeSource
 	payloadSerializer       persistence.PayloadSerializer
 	metricsClient           metrics.Client
+	emitter                 structured.Emitter
 	messagingClient         messaging.Client
 	blobstoreClient         blobstore.Client
 	archivalMetadata        archiver.ArchivalMetadata
@@ -360,6 +362,7 @@ func New(
 		timeSource:              clock.NewRealTimeSource(),
 		payloadSerializer:       persistence.NewPayloadSerializer(),
 		metricsClient:           params.MetricsClient,
+		emitter:                 params.Emitter,
 		messagingClient:         params.MessagingClient,
 		blobstoreClient:         params.BlobstoreClient,
 		archivalMetadata:        params.ArchivalMetadata,
@@ -534,6 +537,11 @@ func (h *Impl) GetMetricsClient() metrics.Client {
 	return h.metricsClient
 }
 
+// GetMetricsEmitter returns a base structured metrics emitter
+func (h *Impl) GetMetricsEmitter() structured.Emitter {
+	return h.emitter
+}
+
 // GetMessagingClient return messaging client
 func (h *Impl) GetMessagingClient() messaging.Client {
 	return h.messagingClient

common/resource/resource_mock.go

@@ -52,6 +52,7 @@ import (
 	"github.com/uber/cadence/common/membership"
 	"github.com/uber/cadence/common/messaging"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	"github.com/uber/cadence/common/mocks"
 	"github.com/uber/cadence/common/persistence"
 	persistenceClient "github.com/uber/cadence/common/persistence/client"
@@ -74,6 +75,7 @@ type (
 		TimeSource              clock.TimeSource
 		PayloadSerializer       persistence.PayloadSerializer
 		MetricsClient           metrics.Client
+		Emitter                 structured.Emitter
 		ArchivalMetadata        *archiver.MockArchivalMetadata
 		ArchiverProvider        *provider.MockArchiverProvider
 		BlobstoreClient         *blobstore.MockClient
@@ -188,6 +190,7 @@ func NewTest(
 		TimeSource:              clock.NewRealTimeSource(),
 		PayloadSerializer:       persistence.NewPayloadSerializer(),
 		MetricsClient:           metrics.NewClient(scope, serviceMetricsIndex),
+		Emitter:                 structured.NewTestEmitter(t, scope),
 		ArchivalMetadata:        &archiver.MockArchivalMetadata{},
 		ArchiverProvider:        &provider.MockArchiverProvider{},
 		BlobstoreClient:         &blobstore.MockClient{},
@@ -295,6 +298,11 @@ func (s *Test) GetMetricsClient() metrics.Client {
 	return s.MetricsClient
 }
 
+// GetMetricsEmitter for testing
+func (s *Test) GetMetricsEmitter() structured.Emitter {
+	return s.Emitter
+}
+
 // GetMessagingClient for testing
 func (s *Test) GetMessagingClient() messaging.Client {
 	panic("user should implement this method for test")

common/resource/resource_test_utils.go

@@ -47,6 +47,7 @@ import (
 	"github.com/uber/cadence/common/membership"
 	"github.com/uber/cadence/common/messaging"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/metrics/structured"
 	"github.com/uber/cadence/common/persistence"
 	persistenceClient "github.com/uber/cadence/common/persistence/client"
 	qrpc "github.com/uber/cadence/common/quotas/global/rpc"
@@ -79,6 +80,7 @@ type Resource interface {
 	GetTimeSource() clock.TimeSource
 	GetPayloadSerializer() persistence.PayloadSerializer
 	GetMetricsClient() metrics.Client
+	GetMetricsEmitter() structured.Emitter
 	GetArchiverProvider() provider.ArchiverProvider
 	GetMessagingClient() messaging.Client
 	GetBlobstoreClient() blobstore.Client

common/resource/types.go

@@ -279,6 +279,7 @@ func NewEngineWithShardContext(
 			historyEngImpl,
 			config,
 			shard.GetMetricsClient(),
+			shard.GetMetricsEmitter(),
 			replicationTaskFetcher,
 			replicationTaskExecutor,
 			shard.GetTimeSource(),