reporting.go

package proxy

import (
	"context"
	"net"
	"strings"
	"time"

	rclient "github.com/go-redis/redis/v8"

	"github.com/getlantern/geo"
	"github.com/getlantern/http-proxy-lantern/v2/common"
	"github.com/getlantern/http-proxy-lantern/v2/listeners"
	"github.com/getlantern/measured"

	"github.com/getlantern/http-proxy-lantern/v2/instrument"
	"github.com/getlantern/http-proxy-lantern/v2/redis"
	"github.com/getlantern/http-proxy-lantern/v2/throttle"
)

var (
	measuredReportingInterval = 1 * time.Minute
)

type reportingConfig struct {
	enabled bool
	wrapper func(ls net.Listener) net.Listener
}

func newReportingConfig(countryLookup geo.CountryLookup, rc *rclient.Client, instrument instrument.Instrument, throttleConfig throttle.Config) *reportingConfig {
	proxiedBytesReporter := func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats, final bool) {
		noDelta := deltaStats.SentTotal == 0 && deltaStats.RecvTotal == 0
		if noDelta && !final {
			// nothing to report on an idle, non-final interval; return before
			// any client IP parsing to keep periodic reporting cheap.
			return
		}

		var client_ip net.IP
		if s, ok := ctx[common.ClientIP].(string); ok {
			client_ip = net.ParseIP(s)
		}

		if final {
			// Record per-session download goodput once at connection close,
			// using the connection's cumulative received bytes and open time.
			// Done before the zero-delta early return below so a session that
			// was idle during its final reporting interval is still counted.
			instrument.SessionGoodput(context.Background(), stats.RecvTotal, stats.Duration, client_ip)
		}

		if noDelta {
			// nothing more to report (final call with no new bytes this interval)
			return
		}
		// Note - sometimes we're missing the platform and version
		platform := fromContext(ctx, common.Platform)
		platformVersion := fromContext(ctx, common.PlatformVersion)
		appVersion := fromContext(ctx, common.AppVersion)
		libraryVersion := fromContext(ctx, common.LibraryVersion)
		app := lowerFromContext(ctx, common.App)
		locale := lowerFromContext(ctx, common.Locale)
		deviceID := fromContext(ctx, common.DeviceID)
		originHost := fromContext(ctx, common.OriginHost)
		probingError := fromContext(ctx, common.ProbingError)
		arch := fromContext(ctx, common.KernelArch)

		dataCapCohort := ""
		throttleSettings, hasThrottleSettings := ctx[common.ThrottleSettings]
		if hasThrottleSettings {
			dataCapCohort = throttleSettings.(*throttle.Settings).Label
		}

		instrument.ProxiedBytes(context.Background(), deltaStats.SentTotal, deltaStats.RecvTotal, platform, platformVersion, libraryVersion, appVersion, app, locale, dataCapCohort, probingError, client_ip, deviceID, originHost, arch)
	}

	var reporter listeners.MeasuredReportFN
	if throttleConfig == nil {
		log.Debug("No throttling configured, don't bother reporting bandwidth usage to Redis")
		reporter = func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats,
			final bool) {
			// noop
		}
	} else if rc != nil {
		reporter = redis.NewMeasuredReporter(countryLookup, rc, measuredReportingInterval, throttleConfig)
	}
	reporter = combineReporter(reporter, proxiedBytesReporter)
	wrapper := func(ls net.Listener) net.Listener {
		return listeners.NewMeasuredListener(ls, measuredReportingInterval, reporter)
	}
	return &reportingConfig{true, wrapper}
}

func fromContext(ctx map[string]interface{}, key string) string {
	value := ctx[key]
	if value != nil {
		return value.(string)
	}
	return ""
}

func lowerFromContext(ctx map[string]interface{}, key string) string {
	return strings.ToLower(fromContext(ctx, key))
}

func combineReporter(reporters ...listeners.MeasuredReportFN) listeners.MeasuredReportFN {
	return func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats, final bool) {
		for _, r := range reporters {
			r(ctx, stats, deltaStats, final)
		}
	}
}