engine: report metrics to opentelemetry.tilt.dev, and give an example…

… of how to report tilt cli commands (#3749)

internal/cli/cli.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/tilt-dev/wmclient/pkg/analytics"
+	"go.opencensus.io/stats"
 
 	"github.com/tilt-dev/tilt/pkg/model"
 
@@ -87,10 +88,17 @@ type tiltCmd interface {
 }
 
 func preCommand(ctx context.Context, cmdName model.TiltSubcommand) (context.Context, func() error) {
-	cleanup := func() error { return nil }
 	l := logger.NewLogger(logLevel(verbose, debug), os.Stdout)
 	ctx = logger.WithLogger(ctx, l)
 
+	ctx, cleanup, err := initMetrics(ctx, cmdName)
+	if err != nil {
+		l.Errorf("Fatal error initializing metrics: %v", err)
+		os.Exit(1)
+	}
+
+	stats.Record(ctx, CommandCountMeasure.M(1))
+
 	a, err := newAnalytics(l, cmdName)
 	if err != nil {
 		l.Errorf("Fatal error initializing analytics: %v", err)

internal/cli/metrics.go

@@ -0,0 +1,65 @@
+package cli
+
+import (
+	"context"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+
+	"github.com/tilt-dev/tilt/internal/engine/metrics"
+	"github.com/tilt-dev/tilt/pkg/model"
+)
+
+var exporter *metrics.DeferredExporter
+var meter view.Meter
+
+// Metric and label names must match the following rules:
+// https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels
+var KeySubCommand = tag.MustNewKey("subcommand")
+
+var CommandCountMeasure = stats.Int64(
+	"cli_count_m",
+	"Number of CLI invocations",
+	stats.UnitDimensionless)
+
+var CommandCount = &view.View{
+	Name:        "cli_count",
+	Measure:     CommandCountMeasure,
+	Description: "Number of CLI invocations",
+	TagKeys:     []tag.Key{KeySubCommand},
+	Aggregation: view.Count(),
+}
+
+func initMetrics(ctx context.Context, cmdName model.TiltSubcommand) (context.Context, func() error, error) {
+	exporter = metrics.NewDeferredExporter()
+	view.RegisterExporter(exporter)
+
+	// TODO(nick): This isn't quite right. Opencensus defaults are really intended
+	// for in-cluster server monitoring, not commandline tools. So we need some
+	// sort of Flush() mechanism to flush the whole reporting pipeline, not just
+	// the exporter.
+	cleanup := func() error {
+		return exporter.Shutdown()
+	}
+
+	err := view.Register(CommandCount)
+	if err != nil {
+		return nil, cleanup, err
+	}
+
+	// In opencensus, we propagate tags with context rather than having
+	// global tags.
+	// https://github.com/census-instrumentation/opencensus-go/issues/786
+	ctx, err = tag.New(ctx,
+		tag.Upsert(KeySubCommand, string(cmdName)))
+	return ctx, cleanup, err
+}
+
+func ProvideDeferredExporter() *metrics.DeferredExporter {
+	return exporter
+}
+
+func ProvideMeter() view.Meter {
+	return meter
+}

internal/cli/wire.go

@@ -31,6 +31,7 @@ import (
 	"github.com/tilt-dev/tilt/internal/engine/k8srollout"
 	"github.com/tilt-dev/tilt/internal/engine/k8swatch"
 	"github.com/tilt-dev/tilt/internal/engine/local"
+	"github.com/tilt-dev/tilt/internal/engine/metrics"
 	"github.com/tilt-dev/tilt/internal/engine/portforward"
 	"github.com/tilt-dev/tilt/internal/engine/runtimelog"
 	"github.com/tilt-dev/tilt/internal/engine/telemetry"
@@ -70,6 +71,8 @@ var BaseWireSet = wire.NewSet(
 
 	docker.SwitchWireSet,
 
+	ProvideDeferredExporter,
+	metrics.NewController,
 	dockercompose.NewDockerComposeClient,
 
 	clockwork.NewRealClock,

internal/engine/metrics/controller.go

@@ -0,0 +1,78 @@
+package metrics
+
+import (
+	"context"
+	"crypto/tls"
+
+	"contrib.go.opencensus.io/exporter/ocagent"
+	"go.opencensus.io/stats/view"
+	"google.golang.org/grpc/credentials"
+
+	"github.com/tilt-dev/tilt/internal/store"
+	"github.com/tilt-dev/tilt/pkg/logger"
+	"github.com/tilt-dev/tilt/pkg/model"
+)
+
+type Controller struct {
+	exporter *DeferredExporter
+	metrics  model.MetricsSettings
+}
+
+func NewController(exporter *DeferredExporter) *Controller {
+	return &Controller{
+		exporter: exporter,
+	}
+}
+
+func (c *Controller) newMetricsSettings(rStore store.RStore) model.MetricsSettings {
+	state := rStore.RLockState()
+	defer rStore.RUnlockState()
+	return state.MetricsSettings
+}
+
+func (c *Controller) OnChange(ctx context.Context, rStore store.RStore) {
+	newMetricsSettings := c.newMetricsSettings(rStore)
+	oldMetricsSettings := c.metrics
+	if newMetricsSettings == oldMetricsSettings {
+		return
+	}
+
+	c.metrics = newMetricsSettings
+	view.SetReportingPeriod(newMetricsSettings.ReportingPeriod)
+
+	if oldMetricsSettings.Enabled && !newMetricsSettings.Enabled {
+		// shutdown the old metrics
+		err := c.exporter.SetRemote(nil)
+		if err != nil {
+			logger.Get(ctx).Debugf("Shutting down metrics: %v", err)
+		}
+	}
+
+	if newMetricsSettings.Enabled {
+		// Replace the existing exporter.
+		options := []ocagent.ExporterOption{
+			ocagent.WithAddress(newMetricsSettings.Address),
+			ocagent.WithServiceName("tilt"),
+		}
+		if newMetricsSettings.Insecure {
+			options = append(options, ocagent.WithInsecure())
+		} else {
+			// default TLS config
+			options = append(options, ocagent.WithTLSCredentials(credentials.NewTLS(&tls.Config{})))
+		}
+		oce, err := ocagent.NewExporter(options...)
+		if err != nil {
+			logger.Get(ctx).Debugf("Creating metrics exporter: %v", err)
+			return
+		}
+
+		err = c.exporter.SetRemote(oce)
+		if err != nil {
+			logger.Get(ctx).Debugf("Setting metrics exporter: %v", err)
+		}
+
+		// TODO(nick): We need a mechanism to synchronously send the existing
+		// aggregates to the remote.
+
+	}
+}

internal/engine/metrics/controller_test.go

@@ -0,0 +1,77 @@
+package metrics
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/tilt-dev/tilt/internal/store"
+	"github.com/tilt-dev/tilt/internal/testutils/tempdir"
+	"github.com/tilt-dev/tilt/pkg/logger"
+	"github.com/tilt-dev/tilt/pkg/model"
+)
+
+func TestEnableMetrics(t *testing.T) {
+	f := newFixture(t)
+
+	assert.Nil(t, f.exp.remote)
+
+	// Verify that enabling metrics creates a remote exporter.
+	ms := model.DefaultMetricsSettings()
+	ms.Enabled = true
+	f.st.SetState(store.EngineState{
+		MetricsSettings: ms,
+	})
+	f.mc.OnChange(f.ctx, f.st)
+
+	remote := f.exp.remote
+	assert.NotNil(t, remote)
+
+	f.mc.OnChange(f.ctx, f.st)
+	assert.Same(t, remote, f.exp.remote)
+
+	// Verify that changing the metrics settings creates a new remote exporter.
+	ms.Insecure = true
+	f.st.SetState(store.EngineState{
+		MetricsSettings: ms,
+	})
+	f.mc.OnChange(f.ctx, f.st)
+	assert.NotSame(t, remote, f.exp.remote)
+
+	// Verify that disabling the metrics settings nulls out the remote exporter.
+	ms.Enabled = false
+	f.st.SetState(store.EngineState{
+		MetricsSettings: ms,
+	})
+	f.mc.OnChange(f.ctx, f.st)
+	assert.Nil(t, f.exp.remote)
+}
+
+type fixture struct {
+	*tempdir.TempDirFixture
+	ctx context.Context
+	st  *store.TestingStore
+	exp *DeferredExporter
+	mc  *Controller
+}
+
+func newFixture(t *testing.T) *fixture {
+	f := tempdir.NewTempDirFixture(t)
+
+	st := store.NewTestingStore()
+
+	l := logger.NewLogger(logger.DebugLvl, os.Stdout)
+	ctx := logger.WithLogger(context.Background(), l)
+
+	exp := NewDeferredExporter()
+	mc := NewController(exp)
+	return &fixture{
+		TempDirFixture: f,
+		ctx:            ctx,
+		st:             st,
+		exp:            exp,
+		mc:             mc,
+	}
+}

internal/engine/metrics/meter.go

@@ -1,6 +1,65 @@
 package metrics
 
 import (
-	_ "contrib.go.opencensus.io/exporter/ocagent"
-	_ "go.opencensus.io/stats/view"
+	"sync"
+
+	"go.opencensus.io/stats/view"
 )
+
+// Glue code between the Tilt subscriber system and the opencensus metrics system.
+
+func NewDeferredExporter() *DeferredExporter {
+	return &DeferredExporter{}
+}
+
+type RemoteExporter interface {
+	view.Exporter
+	Flush()
+	Stop() error
+}
+
+type DeferredExporter struct {
+	mu       sync.Mutex
+	remote   RemoteExporter
+	deferred []*view.Data
+}
+
+func (d *DeferredExporter) Shutdown() error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.remote == nil {
+		return nil
+	}
+	d.remote.Flush()
+	return d.remote.Stop()
+}
+
+func (d *DeferredExporter) SetRemote(remote RemoteExporter) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	oldRemote := d.remote
+	d.remote = remote
+	for _, v := range d.deferred {
+		d.remote.ExportView(v)
+	}
+	d.deferred = nil
+
+	if oldRemote == nil {
+		return nil
+	}
+
+	oldRemote.Flush()
+	return oldRemote.Stop()
+}
+
+func (d *DeferredExporter) ExportView(viewData *view.Data) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if d.remote == nil {
+		d.deferred = append(d.deferred, viewData)
+		return
+	}
+	d.remote.ExportView(viewData)
+}

internal/engine/subscribers.go

@@ -12,6 +12,7 @@ import (
 	"github.com/tilt-dev/tilt/internal/engine/k8srollout"
 	"github.com/tilt-dev/tilt/internal/engine/k8swatch"
 	"github.com/tilt-dev/tilt/internal/engine/local"
+	"github.com/tilt-dev/tilt/internal/engine/metrics"
 	"github.com/tilt-dev/tilt/internal/engine/portforward"
 	"github.com/tilt-dev/tilt/internal/engine/runtimelog"
 	"github.com/tilt-dev/tilt/internal/engine/telemetry"
@@ -47,6 +48,7 @@ func ProvideSubscribers(
 	lc *local.Controller,
 	podm *k8srollout.PodMonitor,
 	ec *exit.Controller,
+	mc *metrics.Controller,
 ) []store.Subscriber {
 	return []store.Subscriber{
 		hud,
@@ -74,5 +76,6 @@ func ProvideSubscribers(
 		lc,
 		podm,
 		ec,
+		mc,
 	}
 }