Include OPA min compat version in telemetry report

This commit extends the telemetry report to include the
minimum compatible version of policies loaded into OPA.
This information can be helpful to get visibility into
era of Rego being adopted in the wild.

Fixes: #6361

Co-authored-by: Stephan Renatus <[email protected]>
Signed-off-by: Ashutosh Narkar <[email protected]>

internal/report/report.go

@@ -10,16 +10,16 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"os"
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/open-policy-agent/opa/keys"
 	"github.com/open-policy-agent/opa/logging"
 
-	"os"
-	"time"
-
 	"github.com/open-policy-agent/opa/plugins/rest"
 	"github.com/open-policy-agent/opa/util"
 	"github.com/open-policy-agent/opa/version"
@@ -38,10 +38,16 @@ var ExternalServiceURL = "https://telemetry.openpolicyagent.org"
 
 // Reporter reports information such as the version, heap usage about the running OPA instance to an external service
 type Reporter struct {
-	body   map[string]string
+	body   map[string]any
 	client rest.Client
+
+	gatherers    map[string]Gatherer
+	gatherersMtx sync.Mutex
 }
 
+// Gatherer represents a mechanism to inject additional data in the telemetry report
+type Gatherer func(ctx context.Context) (any, error)
+
 // DataResponse represents the data returned by the external service
 type DataResponse struct {
 	Latest ReleaseDetails `json:"latest,omitempty"`
@@ -62,8 +68,10 @@ type Options struct {
 
 // New returns an instance of the Reporter
 func New(id string, opts Options) (*Reporter, error) {
-	r := Reporter{}
-	r.body = map[string]string{
+	r := Reporter{
+		gatherers: map[string]Gatherer{},
+	}
+	r.body = map[string]any{
 		"id":      id,
 		"version": version.Version,
 	}
@@ -83,6 +91,9 @@ func New(id string, opts Options) (*Reporter, error) {
 	}
 	r.client = client
 
+	// heap_usage_bytes is always present, so register it unconditionally
+	r.RegisterGatherer("heap_usage_bytes", readRuntimeMemStats)
+
 	return &r, nil
 }
 
@@ -92,9 +103,15 @@ func (r *Reporter) SendReport(ctx context.Context) (*DataResponse, error) {
 	rCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
-	var m runtime.MemStats
-	runtime.ReadMemStats(&m)
-	r.body["heap_usage_bytes"] = strconv.FormatUint(m.Alloc, 10)
+	r.gatherersMtx.Lock()
+	defer r.gatherersMtx.Unlock()
+	for key, g := range r.gatherers {
+		var err error
+		r.body[key], err = g(rCtx)
+		if err != nil {
+			return nil, fmt.Errorf("gather telemetry error for key %s: %w", key, err)
+		}
+	}
 
 	resp, err := r.client.WithJSON(r.body).Do(rCtx, "POST", "/v1/version")
 	if err != nil {
@@ -119,6 +136,12 @@ func (r *Reporter) SendReport(ctx context.Context) (*DataResponse, error) {
 	}
 }
 
+func (r *Reporter) RegisterGatherer(key string, f Gatherer) {
+	r.gatherersMtx.Lock()
+	r.gatherers[key] = f
+	r.gatherersMtx.Unlock()
+}
+
 // IsSet returns true if dr is populated.
 func (dr *DataResponse) IsSet() bool {
 	return dr != nil && dr.Latest.LatestRelease != "" && dr.Latest.Download != "" && dr.Latest.ReleaseNotes != ""
@@ -153,3 +176,9 @@ func (dr *DataResponse) Pretty() string {
 
 	return strings.Join(lines, "\n")
 }
+
+func readRuntimeMemStats(_ context.Context) (any, error) {
+	var m runtime.MemStats
+	runtime.ReadMemStats(&m)
+	return strconv.FormatUint(m.Alloc, 10), nil
+}

internal/report/report_test.go

@@ -114,7 +114,6 @@ func TestReportWithHeapStats(t *testing.T) {
 	}
 
 	_, err = reporter.SendReport(context.Background())
-
 	if err != nil {
 		t.Fatalf("Expected no error but got %v", err)
 	}
@@ -124,6 +123,38 @@ func TestReportWithHeapStats(t *testing.T) {
 	}
 }
 
+func TestReportWithExtraKeys(t *testing.T) {
+
+	// test server
+	baseURL, teardown := getTestServer(nil, http.StatusOK)
+	defer teardown()
+
+	t.Setenv("OPA_TELEMETRY_SERVICE_URL", baseURL)
+
+	reporter, err := New("", Options{})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	reporter.RegisterGatherer("foobear", func(ctx context.Context) (any, error) {
+		return map[string]any{"baz": []string{"one", "two"}}, nil
+	})
+
+	_, err = reporter.SendReport(context.Background())
+	if err != nil {
+		t.Fatalf("Expected no error but got %v", err)
+	}
+
+	if _, ok := reporter.body["foobear"]; !ok {
+		t.Fatal("Expected key \"foobear\" in the report")
+	}
+
+	exp := map[string]any{"baz": []string{"one", "two"}}
+	if act := reporter.body["foobear"]; !reflect.DeepEqual(act, exp) {
+		t.Fatalf("Expected response: %+v but got: %+v", exp, act)
+	}
+}
+
 func TestPretty(t *testing.T) {
 	dr := DataResponse{}
 	resp := dr.Pretty()

plugins/plugins.go

@@ -8,9 +8,11 @@ package plugins
 import (
 	"context"
 	"fmt"
+	mr "math/rand"
 	"sync"
 	"time"
 
+	"github.com/open-policy-agent/opa/internal/report"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.opentelemetry.io/otel/sdk/trace"
 
@@ -148,6 +150,9 @@ const (
 	DefaultTriggerMode TriggerMode = "periodic"
 )
 
+// default interval between OPA report uploads
+var defaultUploadIntervalSec = int64(3600)
+
 // Status has a Plugin's current status plus an optional Message.
 type Status struct {
 	State   State  `json:"state"`
@@ -198,8 +203,13 @@ type Manager struct {
 	tracerProvider               *trace.TracerProvider
 	distributedTacingOpts        tracing.Options
 	registeredNDCacheTriggers    []func(bool)
+	registeredTelemetryGatherers map[string]report.Gatherer
 	bootstrapConfigLabels        map[string]string
 	hooks                        hooks.Hooks
+	enableTelemetry              bool
+	reporter                     *report.Reporter
+	opaReportNotifyCh            chan struct{}
+	stop                         chan chan struct{}
 }
 
 type managerContextKey string
@@ -385,6 +395,21 @@ func WithHooks(hs hooks.Hooks) func(*Manager) {
 	}
 }
 
+// WithEnableTelemetry controls whether OPA will send telemetry reports to an external service.
+func WithEnableTelemetry(enableTelemetry bool) func(*Manager) {
+	return func(m *Manager) {
+		m.enableTelemetry = enableTelemetry
+	}
+}
+
+// WithTelemetryGatherers allows registration of telemetry gatherers which enable injection of additional data in the
+// telemetry report
+func WithTelemetryGatherers(gs map[string]report.Gatherer) func(*Manager) {
+	return func(m *Manager) {
+		m.registeredTelemetryGatherers = gs
+	}
+}
+
 // New creates a new Manager using config.
 func New(raw []byte, id string, store storage.Store, opts ...func(*Manager)) (*Manager, error) {
 
@@ -453,6 +478,27 @@ func New(raw []byte, id string, store storage.Store, opts ...func(*Manager)) (*M
 		return nil, err
 	}
 
+	if m.enableTelemetry {
+		reporter, err := report.New(id, report.Options{Logger: m.logger})
+		if err != nil {
+			return nil, err
+		}
+		m.reporter = reporter
+
+		m.reporter.RegisterGatherer("min_compatible_version", func(_ context.Context) (any, error) {
+			var minimumCompatibleVersion string
+			if m.compiler != nil && m.compiler.Required != nil {
+				minimumCompatibleVersion, _ = m.compiler.Required.MinimumCompatibleVersion()
+			}
+			return minimumCompatibleVersion, nil
+		})
+
+		// register any additional gatherers
+		for k, g := range m.registeredTelemetryGatherers {
+			m.reporter.RegisterGatherer(k, g)
+		}
+	}
+
 	return m, nil
 }
 
@@ -469,6 +515,12 @@ func (m *Manager) Init(ctx context.Context) error {
 		Context: storage.NewContext(),
 	}
 
+	if m.enableTelemetry {
+		m.opaReportNotifyCh = make(chan struct{})
+		m.stop = make(chan chan struct{})
+		go m.sendOPAUpdateLoop(ctx)
+	}
+
 	err := storage.Txn(ctx, m.Store, params, func(txn storage.Transaction) error {
 
 		result, err := initload.InsertAndCompile(ctx, initload.InsertAndCompileOptions{
@@ -497,6 +549,12 @@ func (m *Manager) Init(ctx context.Context) error {
 	})
 
 	if err != nil {
+		if m.stop != nil {
+			done := make(chan struct{})
+			m.stop <- done
+			<-done
+		}
+
 		return err
 	}
 
@@ -673,6 +731,12 @@ func (m *Manager) Stop(ctx context.Context) {
 			m.logger.Error("Error closing store: %v", err)
 		}
 	}
+
+	if m.stop != nil {
+		done := make(chan struct{})
+		m.stop <- done
+		<-done
+	}
 }
 
 // Reconfigure updates the configuration on the manager.
@@ -817,6 +881,11 @@ func (m *Manager) onCommit(ctx context.Context, txn storage.Transaction, event s
 
 	if compiler != nil {
 		m.setCompiler(compiler)
+
+		if m.enableTelemetry && event.PolicyChanged() {
+			m.opaReportNotifyCh <- struct{}{}
+		}
+
 		for _, f := range m.registeredTriggers {
 			f(txn)
 		}
@@ -982,3 +1051,37 @@ func (m *Manager) RegisterNDCacheTrigger(trigger func(bool)) {
 	defer m.mtx.Unlock()
 	m.registeredNDCacheTriggers = append(m.registeredNDCacheTriggers, trigger)
 }
+
+func (m *Manager) sendOPAUpdateLoop(ctx context.Context) {
+	ticker := time.NewTicker(time.Duration(int64(time.Second) * defaultUploadIntervalSec))
+	mr.New(mr.NewSource(time.Now().UnixNano()))
+
+	ctx, cancel := context.WithCancel(ctx)
+
+	var opaReportNotify bool
+
+	for {
+		select {
+		case <-m.opaReportNotifyCh:
+			opaReportNotify = true
+		case <-ticker.C:
+			ticker.Stop()
+
+			if opaReportNotify {
+				opaReportNotify = false
+				_, err := m.reporter.SendReport(ctx)
+				if err != nil {
+					m.logger.WithFields(map[string]interface{}{"err": err}).Debug("Unable to send OPA telemetry report.")
+				}
+			}
+
+			newInterval := mr.Int63n(defaultUploadIntervalSec) + defaultUploadIntervalSec
+			ticker = time.NewTicker(time.Duration(int64(time.Second) * newInterval))
+		case done := <-m.stop:
+			cancel()
+			ticker.Stop()
+			done <- struct{}{}
+			return
+		}
+	}
+}