version metrics (#1199)

Adding a sync.once to base worker that allows user to start reporting version metrics on a per host basis. They can start it once and if desired permanantely stop it on that host.

Author: allenchen2244

internal/internal_worker.go

@@ -51,6 +51,9 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
+var startVersionMetric sync.Once
+var StopMetrics = make(chan struct{})
+
 const (
 	// Set to 2 pollers for now, can adjust later if needed. The typical RTT (round-trip time) is below 1ms within data
 	// center. And the poll API latency is about 5ms. With 2 poller, we could achieve around 300~400 RPS.
@@ -72,6 +75,8 @@ const (
 	defaultMaxConcurrentSessionExecutionSize = 1000 // Large concurrent session execution size (1k)
 
 	testTagsContextKey = "cadence-testTags"
+	clientVersionTag   = "cadence_client_version"
+	clientGauge        = "client_version_metric"
 )
 
 type (
@@ -1255,3 +1260,21 @@ func getTestTags(ctx context.Context) map[string]map[string]string {
 	}
 	return nil
 }
+
+// StartVersionMetrics starts emitting version metrics
+func StartVersionMetrics(metricsScope tally.Scope) {
+	startVersionMetric.Do(func() {
+		go func() {
+			ticker := time.NewTicker(time.Minute)
+			versionTags := map[string]string{clientVersionTag: LibraryVersion}
+			for {
+				select {
+				case <-StopMetrics:
+					return
+				case <-ticker.C:
+					metricsScope.Tagged(versionTags).Gauge(clientGauge).Update(1)
+				}
+			}
+		}()
+	})
+}

internal/internal_worker_base.go

@@ -185,7 +185,6 @@ func newBaseWorker(options baseWorkerOptions, logger *zap.Logger, metricsScope t
 	if options.pollerRate > 0 {
 		bw.pollLimiter = rate.NewLimiter(rate.Limit(options.pollerRate), 1)
 	}
-
 	return bw
 }
 

test/integration_test.go

@@ -24,6 +24,8 @@ package test
 import (
 	"context"
 	"fmt"
+	"github.com/uber-go/tally"
+	"go.uber.org/cadence/internal"
 	"net"
 	"strings"
 	"sync"
@@ -109,12 +111,13 @@ func (ts *IntegrationTestSuite) SetupSuite() {
 		})
 	ts.domainClient = client.NewDomainClient(ts.rpcClient.Interface, &client.Options{})
 	ts.registerDomain()
+	internal.StartVersionMetrics(tally.NoopScope)
 }
 
 func (ts *IntegrationTestSuite) TearDownSuite() {
 	ts.Assertions = require.New(ts.T())
 	ts.rpcClient.Close()
-
+	close(internal.StopMetrics)
 	// allow the pollers to shut down, and ensure there are no goroutine leaks.
 	// this will wait for up to 1 minute for leaks to subside, but exit relatively quickly if possible.
 	max := time.After(time.Minute)