metrics.go

package unleash

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"math"
	"net/http"
	"net/url"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"github.com/Unleash/unleash-go-sdk/v6/internal/api"
	"github.com/Unleash/unleash-go-sdk/v6/internal/impactmetrics"
)

// MetricsData represents the data sent to the unleash server.
type MetricsData struct {
	// AppName is the name of the application.
	AppName string `json:"appName"`

	// InstanceID is the instance identifier.
	InstanceID string `json:"instanceId"`

	// ConnectionId is the connection id for instance.
	ConnectionId string `json:"connectionId"`

	// Bucket is the payload data sent to the server.
	Bucket api.Bucket `json:"bucket"`

	// The runtime version of our Platform
	PlatformVersion string `json:"platformVersion"`

	// The runtime name of our Platform
	PlatformName string `json:"platformName"`

	// Which version of Yggdrasil is being used
	YggdrasilVersion *string `json:"yggdrasilVersion"`

	// Optional field that describes the sdk version (name:version)
	SDKVersion string `json:"sdkVersion"`

	// Which version of the Unleash-Client-Spec is this SDK validated against
	SpecVersion string `json:"specVersion"`

	// ImpactMetrics are optional custom application-level metrics
	ImpactMetrics []impactmetrics.CollectedMetric `json:"impactMetrics,omitempty"`
}

// ClientData represents the data sent to the unleash during registration.
type ClientData struct {
	// AppName is the name of the application.
	AppName string `json:"appName"`

	// InstanceID is the instance identifier.
	InstanceID string `json:"instanceId"`

	// ConnectionId is the connection id for instance.
	ConnectionId string `json:"connectionId"`

	// Optional field that describes the sdk version (name:version)
	SDKVersion string `json:"sdkVersion"`

	// Strategies is a list of names of the strategies supported by the client.
	Strategies []string `json:"strategies"`

	// Started indicates the time at which the client was created.
	Started time.Time `json:"started"`

	// Interval specifies the time interval (in ms) that the client is using for refreshing
	// feature toggles.
	Interval int64 `json:"interval"`

	PlatformVersion string `json:"platformVersion"`

	PlatformName string `json:"platformName"`

	YggdrasilVersion *string `json:"yggdrasilVersion"`

	// Which version of the Unleash-Client-Spec is this SDK validated against
	SpecVersion string `json:"specVersion"`
}

// metricsShutdownTimeout bounds the best-effort final flush performed by
// Close so an unreachable server cannot hang shutdown indefinitely.
const metricsShutdownTimeout = 5 * time.Second

type metric struct {
	// Name is the name of the feature toggle.
	Name string

	// Enabled indicates whether the feature was enabled or not.
	Enabled bool
}

type toggleCounters struct {
	yes int64
	no  int64

	mu       sync.Mutex
	variants map[string]int64
}

type metrics struct {
	metricsChannels
	options        metricsOptions
	started        time.Time
	lastCloseTime  time.Time
	counters       sync.Map // map[string]*toggleCounters
	ticker         *time.Ticker
	close          chan struct{}
	closed         chan struct{}
	ctx            context.Context
	cancel         func()
	maxSkips       float64
	errors         float64
	skips          float64
	metricRegistry impactmetrics.ImpactMetricsDataSource
}

func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
	m := &metrics{
		metricsChannels: channels,
		options:         options,
		started:         time.Now(),
		close:           make(chan struct{}),
		closed:          make(chan struct{}),
		maxSkips:        10,
		errors:          0,
		skips:           0,
		lastCloseTime:   time.Now(),
		metricRegistry:  options.metricRegistry,
	}
	ctx, cancel := context.WithCancel(context.Background())
	m.ctx = ctx
	m.cancel = cancel

	if m.options.httpClient == nil {
		m.options.httpClient = http.DefaultClient
	}

	if m.options.metricsInterval <= 0 {
		m.options.disableMetrics = true
	}
	if !m.options.disableMetrics {
		m.ticker = time.NewTicker(m.options.metricsInterval)
		m.registerInstance()
		go m.sync()
	}

	return m
}

func (m *metrics) Close() error {
	if m.options.disableMetrics {
		return nil
	}
	m.ticker.Stop()
	m.cancel()
	close(m.close)
	<-m.closed
	// Best-effort final flush of anything buffered since the last tick.
	// Bounded so an unreachable server cannot hang Close indefinitely.
	ctx, cancel := context.WithTimeout(context.Background(), metricsShutdownTimeout)
	defer cancel()
	m.flushOnShutdown(ctx)
	return nil
}

func (m *metrics) sync() {
	for {
		select {
		case <-m.ticker.C:
			if m.skips == 0 {
				m.sendMetrics(m.ctx)
			} else {
				m.decrementSkip()
			}
		case <-m.close:
			close(m.closed)
			return
		}
	}
}

func (m *metrics) registerInstance() {
	u, _ := m.options.url.Parse("./client/register")
	payload := m.getClientData()
	resp, err := m.doPost(m.ctx, u, payload)

	if err != nil {
		m.err(err)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusMultipleChoices {
		m.warn(fmt.Errorf("%s return %d", u.String(), resp.StatusCode))
	}

	m.registered <- payload
}
func (m *metrics) backoff() {
	m.errors = math.Min(m.maxSkips, m.errors+1)
	m.skips = m.errors
}

func (m *metrics) configurationError() {
	m.errors = m.maxSkips
	m.skips = m.errors
}

func (m *metrics) successfulPost() {
	m.errors = math.Max(0, m.errors-1)
	m.skips = m.errors
}

func (m *metrics) decrementSkip() {
	m.skips = math.Max(0, m.skips-1)
}

// This does not remove stale toggle names from the map. I don't think there's a safe, lock free way to do that
// The consequence is that if the user archives a lot of toggles this internal representation will not lose those
// toggles until the process is terminated. In practice, I don't believe this is a big problem, just means a
// little bit more memory is held than necessary
func (m *metrics) buildBucketAndReset(lastCloseTime time.Time) (api.Bucket, bool) {
	bucket := api.Bucket{
		Start:   lastCloseTime,
		Toggles: make(map[string]api.ToggleCount),
	}

	m.counters.Range(func(key, value any) bool {
		name := key.(string)
		counter := value.(*toggleCounters)

		yes := atomic.SwapInt64(&counter.yes, 0)
		no := atomic.SwapInt64(&counter.no, 0)

		if yes == 0 && no == 0 {
			counter.mu.Lock()
			emptyVariants := len(counter.variants) == 0
			counter.mu.Unlock()
			if emptyVariants {
				return true
			}
		}

		toggleCounters := api.ToggleCount{
			Yes:      int32(yes),
			No:       int32(no),
			Variants: map[string]int32{},
		}

		// we can have a little locking, as a treat. Variants are likely a luke warm path at best
		// until we have evidence that this is a hot path API, I'd like to keep this simple
		// simple here means a local lock per toggle counter while we swap out the variants map
		counter.mu.Lock()
		if len(counter.variants) > 0 {
			vars := make(map[string]int32, len(counter.variants))
			for vName, cnt := range counter.variants {
				vars[vName] = int32(cnt)
			}
			toggleCounters.Variants = vars

			counter.variants = make(map[string]int64)
		}
		counter.mu.Unlock()

		bucket.Toggles[name] = toggleCounters
		return true
	})

	if len(bucket.Toggles) == 0 {
		return api.Bucket{}, false
	}

	return bucket, true
}

// flushOnShutdown makes a best-effort attempt to POST any buffered toggle
// counts and impact metrics before Close returns. It intentionally emits
// no events: OnSent, OnError, and the backoff counter are all skipped
// because the caller has committed to teardown and cannot react to them.
// The bounded ctx keeps an unreachable server from hanging shutdown.
func (m *metrics) flushOnShutdown(ctx context.Context) {
	bucket, ok := m.buildBucketAndReset(m.lastCloseTime)
	collectedMetrics := impactmetrics.CollectedMetrics(m.metricRegistry.Collect())

	if !ok && collectedMetrics.IsEmpty() {
		return
	}
	bucket.Stop = time.Now()
	payload := MetricsData{
		AppName:          m.options.appName,
		InstanceID:       m.options.instanceId,
		ConnectionId:     m.options.connectionId,
		Bucket:           bucket,
		SDKVersion:       fmt.Sprintf("%s:%s", clientName, clientVersion),
		PlatformName:     "go",
		PlatformVersion:  runtime.Version(),
		YggdrasilVersion: nil,
		SpecVersion:      specVersion,
		ImpactMetrics:    collectedMetrics,
	}

	u, _ := m.options.url.Parse("./client/metrics")
	resp, err := m.doPost(ctx, u, payload)
	if err != nil {
		return
	}
	_ = resp.Body.Close()
}

func (m *metrics) sendMetrics(ctx context.Context) {
	bucket, ok := m.buildBucketAndReset(m.lastCloseTime)
	collectedMetrics := impactmetrics.CollectedMetrics(m.metricRegistry.Collect())

	if !ok && collectedMetrics.IsEmpty() {
		return
	}
	m.lastCloseTime = time.Now()
	bucket.Stop = time.Now()
	payload := MetricsData{
		AppName:          m.options.appName,
		InstanceID:       m.options.instanceId,
		ConnectionId:     m.options.connectionId,
		Bucket:           bucket,
		SDKVersion:       fmt.Sprintf("%s:%s", clientName, clientVersion),
		PlatformName:     "go",
		PlatformVersion:  runtime.Version(),
		YggdrasilVersion: nil,
		SpecVersion:      specVersion,
		ImpactMetrics:    collectedMetrics,
	}

	u, _ := m.options.url.Parse("./client/metrics")
	resp, err := m.doPost(ctx, u, payload)
	if err != nil {
		m.err(err)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusMultipleChoices {
		if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusNotFound {
			m.configurationError()
		} else if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= http.StatusInternalServerError {
			m.backoff()
		}
		m.warn(fmt.Errorf("%s return %d", u.String(), resp.StatusCode))
		// The post failed, re-add the metrics we attempted to send so
		// they are included in the next post.
		m.reinsertBucket(bucket)

		if !collectedMetrics.IsEmpty() {
			m.metricRegistry.Restore(collectedMetrics)
		}

		// Set the start time of the current bucket to the one we
		// attempted to send.
		m.lastCloseTime = bucket.Start

	} else {
		m.successfulPost()
		m.sent <- payload
	}
}

func (m *metrics) doPost(ctx context.Context, url *url.URL, payload interface{}) (*http.Response, error) {
	var body bytes.Buffer
	enc := json.NewEncoder(&body)
	if err := enc.Encode(payload); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", url.String(), &body)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Add("UNLEASH-APPNAME", m.options.appName)
	req.Header.Add("UNLEASH-INSTANCEID", m.options.instanceId)
	req.Header.Add("User-Agent", m.options.appName)
	req.Header.Add("Unleash-Interval", fmt.Sprintf("%d", m.options.metricsInterval.Milliseconds()))

	for k, v := range m.options.headers {
		req.Header[k] = v
	}

	return m.options.httpClient.Do(req)
}

func (m *metrics) getOrCreateCounter(name string) *toggleCounters {
	c, ok := m.counters.Load(name)
	if ok {
		return c.(*toggleCounters)
	}

	nc := &toggleCounters{
		variants: make(map[string]int64),
	}
	actual, _ := m.counters.LoadOrStore(name, nc)
	return actual.(*toggleCounters)
}

func (m *metrics) reinsertBucket(bucket api.Bucket) {
	for name, bucketToggle := range bucket.Toggles {
		counter := m.getOrCreateCounter(name)
		if bucketToggle.Yes != 0 {
			atomic.AddInt64(&counter.yes, int64(bucketToggle.Yes))
		}
		if bucketToggle.No != 0 {
			atomic.AddInt64(&counter.no, int64(bucketToggle.No))
		}

		if len(bucketToggle.Variants) > 0 {
			counter.mu.Lock()
			if counter.variants == nil {
				counter.variants = make(map[string]int64, len(bucketToggle.Variants))
			}
			for vName, cnt := range bucketToggle.Variants {
				if cnt == 0 {
					continue
				}
				counter.variants[vName] += int64(cnt)
			}
			counter.mu.Unlock()
		}
	}
}

func (m *metrics) add(name string, enabled bool, num int32) {
	if m.options.disableMetrics || num == 0 {
		return
	}
	c := m.getOrCreateCounter(name)
	if enabled {
		atomic.AddInt64(&c.yes, int64(num))
	} else {
		atomic.AddInt64(&c.no, int64(num))
	}
}

func (m *metrics) count(name string, enabled bool) {
	if m.options.disableMetrics {
		return
	}
	m.add(name, enabled, 1)
	// best effort delivery, if the channel is full, we skip notifying
	// means under high load we lose some fidelity here, but we avoid blocking
	// the main path. That's probably the correct trade-off. Impression data
	// is the correct insight into this behaviour anyway
	select {
	case m.metricsChannels.count <- metric{Name: name, Enabled: enabled}:
	default:
	}
}

func (m *metrics) countVariants(name string, enabled bool, variantName string) {
	if m.options.disableMetrics {
		return
	}

	m.add(name, enabled, 1)
	// again best effort delivery
	select {
	case m.metricsChannels.count <- metric{Name: name, Enabled: enabled}:
	default:
	}

	c := m.getOrCreateCounter(name)

	c.mu.Lock()
	if c.variants == nil {
		c.variants = make(map[string]int64)
	}
	c.variants[variantName]++
	c.mu.Unlock()
}

func (m *metrics) getClientData() ClientData {
	return ClientData{
		m.options.appName,
		m.options.instanceId,
		m.options.connectionId,
		fmt.Sprintf("%s:%s", clientName, clientVersion),
		m.options.strategies,
		m.started,
		int64(m.options.metricsInterval.Seconds()),
		runtime.Version(),
		"go",
		nil,
		specVersion,
	}
}