https://mattermost.atlassian.net/browse/MM-34932

```release-note
NONE
```
Этот коммит содержится в:
Agniva De Sarker
2021-06-03 12:59:05 +05:30
коммит произвёл GitHub
родитель fff09bf210
Коммит 3299a72adb
776 изменённых файлов: 40081 добавлений и 44702 удалений

8
vendor/github.com/splitio/go-split-commons/v3/conf/conf.go сгенерированный поставляемый
Просмотреть файл

@@ -54,10 +54,11 @@ type TaskPeriods struct {
SplitSync int
SegmentSync int
ImpressionSync int
GaugeSync int
CounterSync int
LatencySync int
GaugeSync int // deprecated
CounterSync int // deprecated
LatencySync int // deprecated
EventsSync int
TelemetrySync int
}
// AdvancedConfig exposes more configurable parameters that can be used to further tailor the sdk to the user's needs
@@ -70,6 +71,7 @@ type AdvancedConfig struct {
SegmentWorkers int
SdkURL string
EventsURL string
TelemetryServiceURL string
EventsBulkSize int64
EventsQueueSize int
ImpressionsQueueSize int

2
vendor/github.com/splitio/go-split-commons/v3/conf/defaults.go сгенерированный поставляемый
Просмотреть файл

@@ -15,6 +15,7 @@ const (
defaultEventsURL = "https://events.split.io/api"
defaultSdkURL = "https://sdk.split.io/api"
defaultStreamingServiceURL = "https://streaming.split.io/sse"
defaultTelemetryServiceURL = "https://telemetry.split.io/api/v1"
)
const (
@@ -48,5 +49,6 @@ func GetDefaultAdvancedConfig() AdvancedConfig {
EventsURL: defaultEventsURL,
SdkURL: defaultSdkURL,
StreamingServiceURL: defaultStreamingServiceURL,
TelemetryServiceURL: defaultTelemetryServiceURL,
}
}

132
vendor/github.com/splitio/go-split-commons/v3/dtos/metrics.go сгенерированный поставляемый
Просмотреть файл

@@ -1,132 +0,0 @@
package dtos
// LatenciesDTO struct mapping latencies post
type LatenciesDTO struct {
MetricName string `json:"name"`
Latencies []int64 `json:"latencies"`
}
// CounterDTO struct mapping counts post
type CounterDTO struct {
MetricName string `json:"name"`
Count int64 `json:"delta"`
}
// GaugeDTO struct mapping gauges post
type GaugeDTO struct {
MetricName string `json:"name"`
Gauge float64 `json:"value"`
}
const maxBuckets = 23
// LatencyDataBulk holds all latencies fetched from storage sorted properly.
type LatencyDataBulk struct {
data map[string]map[string]map[string][]int64
}
// PutLatency adds a new latency to the structure
func (l *LatencyDataBulk) PutLatency(sdk string, machineIP string, metricName string, bucketNumber int, value int64) {
if _, ok := l.data[sdk]; !ok {
l.data[sdk] = make(map[string]map[string][]int64)
}
if _, ok := l.data[sdk][machineIP]; !ok {
l.data[sdk][machineIP] = make(map[string][]int64)
}
if _, ok := l.data[sdk][machineIP][metricName]; !ok {
l.data[sdk][machineIP][metricName] = make([]int64, maxBuckets)
}
l.data[sdk][machineIP][metricName][bucketNumber] = value
}
// ForEach iterates thru all latencies
func (l *LatencyDataBulk) ForEach(callback func(string, string, map[string][]int64)) {
for sdk, byIP := range l.data {
for ip, byName := range byIP {
callback(sdk, ip, byName)
}
}
}
// NewLatencyDataBulk creates a new Latency holding structure
func NewLatencyDataBulk() *LatencyDataBulk {
return &LatencyDataBulk{
data: make(map[string]map[string]map[string][]int64),
}
}
// CounterDataBulk holds all counters fetched from storage sorted properly.
type CounterDataBulk struct {
data map[string]map[string]map[string]int64
}
// PutCounter adds a counter to the structure
func (l *CounterDataBulk) PutCounter(sdk string, machineIP string, metricName string, value int64) {
if _, ok := l.data[sdk]; !ok {
l.data[sdk] = make(map[string]map[string]int64)
}
if _, ok := l.data[sdk][machineIP]; !ok {
l.data[sdk][machineIP] = make(map[string]int64)
}
l.data[sdk][machineIP][metricName] = value
}
// ForEach iterates thru all counters
func (l *CounterDataBulk) ForEach(callback func(string, string, map[string]int64)) {
for sdk, byIP := range l.data {
for ip, byName := range byIP {
callback(sdk, ip, byName)
}
}
}
// NewCounterDataBulk creates a new Counter holding structure
func NewCounterDataBulk() *CounterDataBulk {
return &CounterDataBulk{
data: make(map[string]map[string]map[string]int64),
}
}
// GaugeDataBulk holds all gauges fetched from storage sorted properly.
type GaugeDataBulk struct {
data map[string]map[string]map[string]float64
}
// PutGauge adds a gauge to the structure
func (l *GaugeDataBulk) PutGauge(sdk string, machineIP string, metricName string, value float64) {
if _, ok := l.data[sdk]; !ok {
l.data[sdk] = make(map[string]map[string]float64)
}
if _, ok := l.data[sdk][machineIP]; !ok {
l.data[sdk][machineIP] = make(map[string]float64)
}
l.data[sdk][machineIP][metricName] = value
}
// ForEach iterates thru all gauges
func (l *GaugeDataBulk) ForEach(callback func(string, string, string, float64)) {
for sdk, byIP := range l.data {
for ip, byName := range byIP {
for name, value := range byName {
callback(sdk, ip, name, value)
}
}
}
}
// NewGaugeDataBulk creates a new Gauge holding structure
func NewGaugeDataBulk() *GaugeDataBulk {
return &GaugeDataBulk{
data: make(map[string]map[string]map[string]float64),
}
}

126
vendor/github.com/splitio/go-split-commons/v3/dtos/telemetry.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,126 @@
package dtos
// LastSynchronization struct
type LastSynchronization struct {
Splits int64 `json:"sp,omitempty"`
Segments int64 `json:"se,omitempty"`
Impressions int64 `json:"im,omitempty"`
ImpressionsCount int64 `json:"ic,omitempty"`
Events int64 `json:"ev,omitempty"`
Token int64 `json:"to,omitempty"`
Telemetry int64 `json:"te,omitempty"`
}
// HTTPErrors struct
type HTTPErrors struct {
Splits map[int]int64 `json:"sp,omitempty"`
Segments map[int]int64 `json:"se,omitempty"`
Impressions map[int]int64 `json:"im,omitempty"`
ImpressionsCount map[int]int64 `json:"ic,omitempty"`
Events map[int]int64 `json:"ev,omitempty"`
Token map[int]int64 `json:"to,omitempty"`
Telemetry map[int]int64 `json:"te,omitempty"`
}
// HTTPLatencies struct
type HTTPLatencies struct {
Splits []int64 `json:"sp,omitempty"`
Segments []int64 `json:"se,omitempty"`
Impressions []int64 `json:"im,omitempty"`
ImpressionsCount []int64 `json:"ic,omitempty"`
Events []int64 `json:"ev,omitempty"`
Token []int64 `json:"to,omitempty"`
Telemetry []int64 `json:"te,omitempty"`
}
// MethodLatencies struct
type MethodLatencies struct {
Treatment []int64 `json:"t,omitempty"`
Treatments []int64 `json:"ts,omitempty"`
TreatmentWithConfig []int64 `json:"tc,omitempty"`
TreatmentsWithConfig []int64 `json:"tcs,omitempty"`
Track []int64 `json:"tr,omitempty"`
}
// MethodExceptions struct
type MethodExceptions struct {
Treatment int64 `json:"t,omitempty"`
Treatments int64 `json:"ts,omitempty"`
TreatmentWithConfig int64 `json:"tc,omitempty"`
TreatmentsWithConfig int64 `json:"tcs,omitempty"`
Track int64 `json:"tr,omitempty"`
}
// StreamingEvent struct
type StreamingEvent struct {
Type int `json:"e,omitempty"`
Data int64 `json:"d,omitempty"`
Timestamp int64 `json:"t,omitempty"`
}
// TelemetryQueueObject struct mapping telemetry
type TelemetryQueueObject struct {
Metadata Metadata `json:"m"`
Config Config `json:"t"`
}
// Rates struct
type Rates struct {
Splits int64 `json:"sp,omitempty"`
Segments int64 `json:"se,omitempty"`
Impressions int64 `json:"im,omitempty"`
Events int64 `json:"ev,omitempty"`
Telemetry int64 `json:"te,omitempty"`
}
// URLOverrides struct
type URLOverrides struct {
Sdk bool `json:"s,omitempty"`
Events bool `json:"e,omitempty"`
Auth bool `json:"a,omitempty"`
Stream bool `json:"st,omitempty"`
Telemetry bool `json:"t,omitempty"`
}
// Config data for initial configs metrics
type Config struct {
OperationMode int `json:"oM,omitempty"`
StreamingEnabled bool `json:"sE,omitempty"`
Storage string `json:"st,omitempty"`
Rates *Rates `json:"rR,omitempty"`
URLOverrides *URLOverrides `json:"uO,omitempty"`
ImpressionsQueueSize int64 `json:"iQ,omitempty"`
EventsQueueSize int64 `json:"eQ,omitempty"`
ImpressionsMode int `json:"iM,omitempty"`
ImpressionsListenerEnabled bool `json:"iL,omitempty"`
HTTPProxyDetected bool `json:"hP,omitempty"`
ActiveFactories int64 `json:"aF,omitempty"`
RedundantFactories int64 `json:"rF,omitempty"`
TimeUntilReady int64 `json:"tR,omitempty"`
BurTimeouts int64 `json:"bT,omitempty"`
NonReadyUsages int64 `json:"nR,omitempty"`
Integrations []string `json:"i,omitempty"`
Tags []string `json:"t,omitempty"`
}
// Stats data sent by sdks pereiodically
type Stats struct {
LastSynchronizations *LastSynchronization `json:"lS,omitempty"`
MethodLatencies *MethodLatencies `json:"mL,omitempty"`
MethodExceptions *MethodExceptions `json:"mE,omitempty"`
HTTPErrors *HTTPErrors `json:"hE,omitempty"`
HTTPLatencies *HTTPLatencies `json:"hL,omitempty"`
TokenRefreshes int64 `json:"tR,omitempty"`
AuthRejections int64 `json:"aR,omitempty"`
ImpressionsQueued int64 `json:"iQ,omitempty"`
ImpressionsDeduped int64 `json:"iDe,omitempty"`
ImpressionsDropped int64 `json:"iDr,omitempty"`
SplitCount int64 `json:"spC,omitempty"`
SegmentCount int64 `json:"seC,omitempty"`
SegmentKeyCount int64 `json:"skC,omitempty"`
SessionLengthMs int64 `json:"sL,omitempty"`
EventsQueued int64 `json:"eQ,omitempty"`
EventsDropped int64 `json:"eD,omitempty"`
StreamingEvents []StreamingEvent `json:"sE,omitempty"`
Tags []string `json:"t,omitempty"`
}

7
vendor/github.com/splitio/go-split-commons/v3/provisional/impmanager.go сгенерированный поставляемый
Просмотреть файл

@@ -5,6 +5,8 @@ import (
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-split-commons/v3/util"
)
@@ -22,10 +24,11 @@ type ImpressionManagerImpl struct {
shouldAddPreviousTime bool
isOptimized bool
listenerEnabled bool
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewImpressionManager creates new ImpManager
func NewImpressionManager(managerConfig conf.ManagerConfig, impressionCounter *ImpressionsCounter) (ImpressionManager, error) {
func NewImpressionManager(managerConfig conf.ManagerConfig, impressionCounter *ImpressionsCounter, runtimeTelemetry storage.TelemetryRuntimeProducer) (ImpressionManager, error) {
impressionObserver, err := NewImpressionObserver(lastSeenCacheSize)
if err != nil {
return nil, err
@@ -37,6 +40,7 @@ func NewImpressionManager(managerConfig conf.ManagerConfig, impressionCounter *I
shouldAddPreviousTime: util.ShouldAddPreviousTime(managerConfig),
isOptimized: impressionCounter != nil && util.ShouldBeOptimized(managerConfig),
listenerEnabled: managerConfig.ListenerEnabled,
runtimeTelemetry: runtimeTelemetry,
}
return impManager, nil
@@ -72,5 +76,6 @@ func (i *ImpressionManagerImpl) ProcessImpressions(impressions []dtos.Impression
forLog, forListener = i.processImpression(impression, forLog, forListener)
}
i.runtimeTelemetry.RecordImpressionsStats(telemetry.ImpressionsDeduped, int64(len(impressions)-len(forLog)))
return forLog, forListener
}

14
vendor/github.com/splitio/go-split-commons/v3/push/constants.go сгенерированный поставляемый
Просмотреть файл

@@ -1,14 +0,0 @@
package push
const (
workerStatusIdle = iota
workerStatusRunning
workerStatusShuttingDown
)
const (
pushManagerStatusIdle = iota
pushManagerStatusInitializing
pushManagerStatusRunning
pushManagerStatusShuttingDown
)

46
vendor/github.com/splitio/go-split-commons/v3/push/manager.go сгенерированный поставляемый
Просмотреть файл

@@ -11,6 +11,8 @@ import (
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/service/api/sse"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/common"
"github.com/splitio/go-toolkit/v4/logging"
"github.com/splitio/go-toolkit/v4/struct/traits/lifecycle"
@@ -48,13 +50,9 @@ type ManagerImpl struct {
feedback FeedbackLoop
nextRefresh *time.Timer
refreshTokenMutex sync.Mutex
/*
running *gtSync.AtomicBool
status int32
shutdownWaiter chan struct{}
*/
lifecycle lifecycle.Manager
logger logging.LoggerInterface
lifecycle lifecycle.Manager
logger logging.LoggerInterface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// FeedbackLoop is a type alias for the type of chan that must be supplied for push status tobe propagated
@@ -67,6 +65,9 @@ func NewManager(
cfg *conf.AdvancedConfig,
feedbackLoop chan<- int64,
authAPI service.AuthClient,
runtimeTelemetry storage.TelemetryRuntimeProducer,
metadata dtos.Metadata,
clientKey *string,
) (*ManagerImpl, error) {
processor, err := NewProcessor(cfg.SplitUpdateQueueSize, cfg.SegmentUpdateQueueSize, synchronizer, logger)
@@ -74,7 +75,7 @@ func NewManager(
return nil, fmt.Errorf("error instantiating processor: %w", err)
}
statusTracker := NewStatusTracker(logger)
statusTracker := NewStatusTracker(logger, runtimeTelemetry)
parser := &NotificationParserImpl{
logger: logger,
onSplitUpdate: processor.ProcessSplitChangeUpdate,
@@ -86,13 +87,14 @@ func NewManager(
}
manager := &ManagerImpl{
authAPI: authAPI,
sseClient: sse.NewStreamingClient(cfg, logger),
statusTracker: statusTracker,
feedback: feedbackLoop,
processor: processor,
parser: parser,
logger: logger,
authAPI: authAPI,
sseClient: sse.NewStreamingClient(cfg, logger, metadata, clientKey),
statusTracker: statusTracker,
feedback: feedbackLoop,
processor: processor,
parser: parser,
logger: logger,
runtimeTelemetry: runtimeTelemetry,
}
manager.lifecycle.Setup()
return manager, nil
@@ -135,21 +137,29 @@ func (m *ManagerImpl) StopWorkers() {
}
func (m *ManagerImpl) performAuthentication() (*dtos.Token, *int64) {
before := time.Now()
token, err := m.authAPI.Authenticate()
if err != nil {
if errType, ok := err.(dtos.HTTPError); ok {
m.runtimeTelemetry.RecordSyncError(telemetry.TokenSync, errType.Code)
if errType.Code >= http.StatusInternalServerError {
m.logger.Error(fmt.Sprintf("Error authenticating: %s", err.Error()))
return nil, common.Int64Ref(StatusRetryableError)
}
if errType.Code == http.StatusUnauthorized {
m.runtimeTelemetry.RecordAuthRejections() // Only 401
}
return nil, common.Int64Ref(StatusNonRetryableError) // 400, 401, etc
}
// Not an HTTP eerror, most likely a tcp/bad connection. Should retry
// Not an HTTP error, most likely a tcp/bad connection. Should retry
return nil, common.Int64Ref(StatusRetryableError)
}
m.runtimeTelemetry.RecordSyncLatency(telemetry.TokenSync, time.Since(before).Nanoseconds())
if !token.PushEnabled {
return nil, common.Int64Ref(StatusNonRetryableError)
}
m.runtimeTelemetry.RecordTokenRefreshes()
m.runtimeTelemetry.RecordSuccessfulSync(telemetry.TokenSync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return token, nil
}
@@ -199,6 +209,8 @@ func (m *ManagerImpl) triggerConnectionFlow() {
m.logger.Warning("Failed to calculate next token expiration time. Defaulting to 50 minutes")
when = 50 * time.Minute
}
// Tracking TOKEN_REFRESHES
m.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeTokenRefresh, when.Milliseconds()))
m.withRefreshTokenLock(func() {
m.nextRefresh = time.AfterFunc(when, func() {
m.logger.Info("Refreshing SSE auth token.")
@@ -206,6 +218,8 @@ func (m *ManagerImpl) triggerConnectionFlow() {
m.Start()
})
})
// Tracking CONNECTION_ESTABLISHED
m.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeSSEConnectionEstablished, 0))
m.feedback <- StatusUp
case sse.StatusConnectionFailed:
m.lifecycle.AbnormalShutdown()

26
vendor/github.com/splitio/go-split-commons/v3/push/statustracker.go сгенерированный поставляемый
Просмотреть файл

@@ -4,10 +4,17 @@ import (
"fmt"
"sync"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/common"
"github.com/splitio/go-toolkit/v4/logging"
)
const (
pri = "control_pri"
sec = "control_sec"
)
// StatusTracker keeps track of the status of the push subsystem and generates appropriate status change notifications.
type StatusTracker interface {
HandleOccupancy(*OccupancyMessage) *int64
@@ -28,6 +35,7 @@ type StatusTrackerImpl struct {
lastControlMessage string
lastStatusPropagated int64
shutdownExpected bool
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NotifySSEShutdownExpected should be called when we are forcefully closing the SSE client
@@ -41,7 +49,7 @@ func (p *StatusTrackerImpl) NotifySSEShutdownExpected() {
func (p *StatusTrackerImpl) Reset() {
p.mutex.Lock()
defer p.mutex.Unlock()
p.occupancy = map[string]int64{"control_pri": 2, "control_sec": 2}
p.occupancy = map[string]int64{pri: 2, sec: 2}
p.lastControlMessage = ControlTypeStreamingEnabled
p.lastStatusPropagated = StatusUp
p.shutdownExpected = false
@@ -63,6 +71,13 @@ func (p *StatusTrackerImpl) HandleOccupancy(message *OccupancyMessage) (newStatu
p.lastOccupancyTimestamp = message.Timestamp()
p.occupancy[channel] = message.Publishers()
// Tracking OccupancyEvent
switch channel {
case pri:
p.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeOccupancyPri, message.Publishers()))
case sec:
p.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeOccupancySec, message.Publishers()))
}
return p.updateStatus()
}
@@ -77,6 +92,9 @@ func (p *StatusTrackerImpl) HandleAblyError(errorEvent *AblyError) (newStatus *i
// Regardless of whether the error is retryable or not, we're going to close the connection
p.shutdownExpected = true
// Tracking ABLY_ERROR
p.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeAblyError, int64(errorEvent.Code())))
if errorEvent.IsRetryable() {
p.logger.Info("Received retryable error message. Restarting SSE connection with backoff")
return p.propagateStatus(StatusRetryableError)
@@ -109,14 +127,16 @@ func (p *StatusTrackerImpl) HandleDisconnection() *int64 {
p.mutex.Lock()
defer p.mutex.Unlock()
if !p.shutdownExpected {
p.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeConnectionError, telemetry.NonRequested))
return p.propagateStatus(StatusRetryableError)
}
p.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeConnectionError, telemetry.Requested))
return nil
}
// NewStatusTracker returns a new StatusTracker
func NewStatusTracker(logger logging.LoggerInterface) *StatusTrackerImpl {
tracker := &StatusTrackerImpl{logger: logger}
func NewStatusTracker(logger logging.LoggerInterface, runtimeTelemetry storage.TelemetryRuntimeProducer) *StatusTrackerImpl {
tracker := &StatusTrackerImpl{logger: logger, runtimeTelemetry: runtimeTelemetry}
tracker.Reset()
return tracker
}

10
vendor/github.com/splitio/go-split-commons/v3/service/api/client.go сгенерированный поставляемый
Просмотреть файл

@@ -30,7 +30,6 @@ type Client interface {
type HTTPClient struct {
url string
httpClient *http.Client
headers map[string]string
logger logging.LoggerInterface
apikey string
metadata dtos.Metadata
@@ -44,8 +43,7 @@ func NewHTTPClient(
logger logging.LoggerInterface,
metadata dtos.Metadata,
) Client {
var timeout int
timeout = cfg.HTTPTimeout
timeout := cfg.HTTPTimeout
client := &http.Client{Timeout: time.Duration(timeout) * time.Second}
return &HTTPClient{
url: endpoint,
@@ -66,11 +64,9 @@ func (c *HTTPClient) Get(service string, headers map[string]string) ([]byte, err
c.logger.Debug("Authorization [ApiKey]: ", logging.ObfuscateAPIKey(authorization))
req.Header.Add("Accept-Encoding", "gzip")
req.Header.Add("Content-Type", "application/json")
req.Header.Add("SplitSDKVersion", c.metadata.SDKVersion)
req.Header.Add("SplitSDKMachineName", c.metadata.MachineName)
req.Header.Add("SplitSDKMachineIP", c.metadata.MachineIP)
parsedHeaders := AddMetadataToHeaders(c.metadata, headers, nil)
for headerName, headerValue := range headers {
for headerName, headerValue := range parsedHeaders {
req.Header.Add(headerName, headerValue)
}

32
vendor/github.com/splitio/go-split-commons/v3/service/api/helpers.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,32 @@
package api
import "github.com/splitio/go-split-commons/v3/dtos"
const (
splitSDKVersion = "SplitSDKVersion"
splitSDKMachineName = "SplitSDKMachineName"
splitSDKMachineIP = "SplitSDKMachineIP"
splitSDKClientKey = "SplitSDKClientKey"
unknown = "unknown"
na = "NA"
)
// AddMetadataToHeaders adds metadata in headers
func AddMetadataToHeaders(metadata dtos.Metadata, extraHeaders map[string]string, clientKey *string) map[string]string {
headers := make(map[string]string)
headers[splitSDKVersion] = metadata.SDKVersion
if metadata.MachineName != na && metadata.MachineName != unknown {
headers[splitSDKMachineName] = metadata.MachineName
}
if metadata.MachineIP != na && metadata.MachineIP != unknown {
headers[splitSDKMachineIP] = metadata.MachineIP
}
for header, value := range extraHeaders {
headers[header] = value
}
if clientKey != nil {
headers[splitSDKClientKey] = *clientKey
}
return headers
}

15
vendor/github.com/splitio/go-split-commons/v3/service/api/http_fetchers.go сгенерированный поставляемый
Просмотреть файл

@@ -7,6 +7,7 @@ import (
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -41,12 +42,7 @@ type HTTPSplitFetcher struct {
}
// NewHTTPSplitFetcher instantiates and return an HTTPSplitFetcher
func NewHTTPSplitFetcher(
apikey string,
cfg conf.AdvancedConfig,
logger logging.LoggerInterface,
metadata dtos.Metadata,
) *HTTPSplitFetcher {
func NewHTTPSplitFetcher(apikey string, cfg conf.AdvancedConfig, logger logging.LoggerInterface, metadata dtos.Metadata) service.SplitFetcher {
return &HTTPSplitFetcher{
httpFetcherBase: httpFetcherBase{
client: NewHTTPClient(apikey, cfg, cfg.SdkURL, logger, metadata),
@@ -79,12 +75,7 @@ type HTTPSegmentFetcher struct {
}
// NewHTTPSegmentFetcher instantiates and returns a new HTTPSegmentFetcher.
func NewHTTPSegmentFetcher(
apikey string,
cfg conf.AdvancedConfig,
logger logging.LoggerInterface,
metadata dtos.Metadata,
) *HTTPSegmentFetcher {
func NewHTTPSegmentFetcher(apikey string, cfg conf.AdvancedConfig, logger logging.LoggerInterface, metadata dtos.Metadata) service.SegmentFetcher {
return &HTTPSegmentFetcher{
httpFetcherBase: httpFetcherBase{
client: NewHTTPClient(apikey, cfg, cfg.SdkURL, logger, metadata),

149
vendor/github.com/splitio/go-split-commons/v3/service/api/http_recorders.go сгенерированный поставляемый
Просмотреть файл

@@ -5,6 +5,7 @@ import (
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -15,20 +16,7 @@ type httpRecorderBase struct {
// RecordRaw records raw data
func (h *httpRecorderBase) RecordRaw(url string, data []byte, metadata dtos.Metadata, extraHeaders map[string]string) error {
headers := make(map[string]string)
headers["SplitSDKVersion"] = metadata.SDKVersion
if metadata.MachineName != "NA" && metadata.MachineName != "unknown" {
headers["SplitSDKMachineName"] = metadata.MachineName
}
if metadata.MachineIP != "NA" && metadata.MachineIP != "unknown" {
headers["SplitSDKMachineIP"] = metadata.MachineIP
}
if extraHeaders != nil {
for header, value := range extraHeaders {
headers[header] = value
}
}
return h.client.Post(url, data, headers)
return h.client.Post(url, data, AddMetadataToHeaders(metadata, extraHeaders, nil))
}
// HTTPImpressionRecorder is a struct responsible for submitting impression bulks to the backend
@@ -74,11 +62,7 @@ func (i *HTTPImpressionRecorder) RecordImpressionsCount(pf dtos.ImpressionsCount
}
// NewHTTPImpressionRecorder instantiates an HTTPImpressionRecorder
func NewHTTPImpressionRecorder(
apikey string,
cfg conf.AdvancedConfig,
logger logging.LoggerInterface,
) *HTTPImpressionRecorder {
func NewHTTPImpressionRecorder(apikey string, cfg conf.AdvancedConfig, logger logging.LoggerInterface) service.ImpressionsRecorder {
client := NewHTTPClient(apikey, cfg, cfg.EventsURL, logger, dtos.Metadata{})
return &HTTPImpressionRecorder{
httpRecorderBase: httpRecorderBase{
@@ -88,77 +72,6 @@ func NewHTTPImpressionRecorder(
}
}
// HTTPMetricsRecorder is a struct responsible for submitting metrics (latency, gauge, counters) to the backend
type HTTPMetricsRecorder struct {
httpRecorderBase
}
// RecordCounters method submits counter metrics to the backend
func (m *HTTPMetricsRecorder) RecordCounters(counters []dtos.CounterDTO, metadata dtos.Metadata) error {
data, err := json.Marshal(counters)
if err != nil {
m.logger.Error("Error marshaling JSON", err.Error())
return err
}
err = m.RecordRaw("/metrics/counters", data, metadata, nil)
if err != nil {
m.logger.Error("Error posting counters", err.Error())
return err
}
return nil
}
// RecordLatencies method submits latency metrics to the backend
func (m *HTTPMetricsRecorder) RecordLatencies(latencies []dtos.LatenciesDTO, metadata dtos.Metadata) error {
data, err := json.Marshal(latencies)
if err != nil {
m.logger.Error("Error marshaling JSON", err.Error())
return err
}
err = m.RecordRaw("/metrics/times", data, metadata, nil)
if err != nil {
m.logger.Error("Error posting latencies", err.Error())
return err
}
return nil
}
// RecordGauge method submits gauge metrics to the backend
func (m *HTTPMetricsRecorder) RecordGauge(gauge dtos.GaugeDTO, metadata dtos.Metadata) error {
data, err := json.Marshal(gauge)
if err != nil {
m.logger.Error("Error marshaling JSON", err.Error())
return err
}
err = m.RecordRaw("/metrics/gauge", data, metadata, nil)
if err != nil {
m.logger.Error("Error posting gauges", err.Error())
return err
}
return nil
}
// NewHTTPMetricsRecorder instantiates an HTTPMetricsRecorder
func NewHTTPMetricsRecorder(
apikey string,
cfg conf.AdvancedConfig,
logger logging.LoggerInterface,
) *HTTPMetricsRecorder {
client := NewHTTPClient(apikey, cfg, cfg.EventsURL, logger, dtos.Metadata{})
return &HTTPMetricsRecorder{
httpRecorderBase: httpRecorderBase{
client: client,
logger: logger,
},
}
}
// HTTPEventsRecorder is a struct responsible for submitting events bulks to the backend
type HTTPEventsRecorder struct {
httpRecorderBase
@@ -182,11 +95,7 @@ func (i *HTTPEventsRecorder) Record(events []dtos.EventDTO, metadata dtos.Metada
}
// NewHTTPEventsRecorder instantiates an HTTPEventsRecorder
func NewHTTPEventsRecorder(
apikey string,
cfg conf.AdvancedConfig,
logger logging.LoggerInterface,
) *HTTPEventsRecorder {
func NewHTTPEventsRecorder(apikey string, cfg conf.AdvancedConfig, logger logging.LoggerInterface) service.EventsRecorder {
client := NewHTTPClient(apikey, cfg, cfg.EventsURL, logger, dtos.Metadata{})
return &HTTPEventsRecorder{
httpRecorderBase: httpRecorderBase{
@@ -195,3 +104,53 @@ func NewHTTPEventsRecorder(
},
}
}
// HTTPTelemetryRecorder is a struct responsible for submitting telemetry to the backend
type HTTPTelemetryRecorder struct {
httpRecorderBase
}
// NewHTTPTelemetryRecorder instantiates an HTTPTelemetryRecorder
func NewHTTPTelemetryRecorder(apikey string, cfg conf.AdvancedConfig, logger logging.LoggerInterface) service.TelemetryRecorder {
client := NewHTTPClient(apikey, cfg, cfg.TelemetryServiceURL, logger, dtos.Metadata{})
return &HTTPTelemetryRecorder{
httpRecorderBase: httpRecorderBase{
client: client,
logger: logger,
},
}
}
// RecordConfig method submits config
func (m *HTTPTelemetryRecorder) RecordConfig(config dtos.Config, metadata dtos.Metadata) error {
data, err := json.Marshal(config)
if err != nil {
m.logger.Error("Error marshaling JSON", err.Error())
return err
}
err = m.RecordRaw("/metrics/config", data, metadata, nil)
if err != nil {
m.logger.Error("Error posting config", err.Error())
return err
}
return nil
}
// RecordStats method submits stats
func (m *HTTPTelemetryRecorder) RecordStats(stats dtos.Stats, metadata dtos.Metadata) error {
data, err := json.Marshal(stats)
if err != nil {
m.logger.Error("Error marshaling JSON", err.Error())
return err
}
err = m.RecordRaw("/metrics/usage", data, metadata, nil)
if err != nil {
m.logger.Error("Error posting usage", err.Error())
return err
}
return nil
}

35
vendor/github.com/splitio/go-split-commons/v3/service/api/split.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,35 @@
package api
import (
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-toolkit/v4/logging"
)
// SplitAPI struct for fetchers and recorders
type SplitAPI struct {
AuthClient service.AuthClient
SplitFetcher service.SplitFetcher
SegmentFetcher service.SegmentFetcher
ImpressionRecorder service.ImpressionsRecorder
EventRecorder service.EventsRecorder
TelemetryRecorder service.TelemetryRecorder
}
// NewSplitAPI creates new splitAPI
func NewSplitAPI(
apikey string,
conf conf.AdvancedConfig,
logger logging.LoggerInterface,
metadata dtos.Metadata,
) *SplitAPI {
return &SplitAPI{
AuthClient: NewAuthAPIClient(apikey, conf, logger, metadata),
SplitFetcher: NewHTTPSplitFetcher(apikey, conf, logger, metadata),
SegmentFetcher: NewHTTPSegmentFetcher(apikey, conf, logger, metadata),
ImpressionRecorder: NewHTTPImpressionRecorder(apikey, conf, logger),
EventRecorder: NewHTTPEventsRecorder(apikey, conf, logger),
TelemetryRecorder: NewHTTPTelemetryRecorder(apikey, conf, logger),
}
}

10
vendor/github.com/splitio/go-split-commons/v3/service/api/sse/client.go сгенерированный поставляемый
Просмотреть файл

@@ -5,6 +5,8 @@ import (
"strings"
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service/api"
"github.com/splitio/go-toolkit/v4/logging"
"github.com/splitio/go-toolkit/v4/sse"
"github.com/splitio/go-toolkit/v4/struct/traits/lifecycle"
@@ -28,6 +30,8 @@ type StreamingClientImpl struct {
sseClient *sse.Client
logger logging.LoggerInterface
lifecycle lifecycle.Manager
metadata dtos.Metadata
clientKey *string
}
// Status constants
@@ -42,12 +46,14 @@ const (
type IncomingMessage = sse.RawEvent
// NewStreamingClient creates new SSE Client
func NewStreamingClient(cfg *conf.AdvancedConfig, logger logging.LoggerInterface) *StreamingClientImpl {
func NewStreamingClient(cfg *conf.AdvancedConfig, logger logging.LoggerInterface, metadata dtos.Metadata, clientKey *string) *StreamingClientImpl {
sseClient, _ := sse.NewClient(cfg.StreamingServiceURL, keepAlive, logger)
client := &StreamingClientImpl{
sseClient: sseClient,
logger: logger,
metadata: metadata,
clientKey: clientKey,
}
client.lifecycle.Setup()
return client
@@ -72,7 +78,7 @@ func (s *StreamingClientImpl) ConnectStreaming(token string, streamingStatus cha
return
}
firstEventReceived := gtSync.NewAtomicBool(false)
out := s.sseClient.Do(params, func(m IncomingMessage) {
out := s.sseClient.Do(params, api.AddMetadataToHeaders(s.metadata, nil, s.clientKey), func(m IncomingMessage) {
if firstEventReceived.TestAndSet() && !m.IsError() {
streamingStatus <- StatusFirstEventOk
}

9
vendor/github.com/splitio/go-split-commons/v3/service/interfaces.go сгенерированный поставляемый
Просмотреть файл

@@ -25,11 +25,10 @@ type ImpressionsRecorder interface {
RecordImpressionsCount(pf dtos.ImpressionsCountDTO, metadata dtos.Metadata) error
}
// MetricsRecorder interface to be implemented by Metrics loggers
type MetricsRecorder interface {
RecordLatencies(latencies []dtos.LatenciesDTO, metadata dtos.Metadata) error
RecordCounters(counters []dtos.CounterDTO, metadata dtos.Metadata) error
RecordGauge(gauge dtos.GaugeDTO, metadata dtos.Metadata) error
// TelemetryRecorder interface to be implemented by Telemetry loggers
type TelemetryRecorder interface {
RecordConfig(config dtos.Config, metadata dtos.Metadata) error
RecordStats(stats dtos.Stats, metadata dtos.Metadata) error
}
// EventsRecorder interface to post events

35
vendor/github.com/splitio/go-split-commons/v3/service/split.go сгенерированный поставляемый
Просмотреть файл

@@ -1,35 +0,0 @@
package service
import (
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service/api"
"github.com/splitio/go-toolkit/v4/logging"
)
// SplitAPI struct for fetchers and recorders
type SplitAPI struct {
AuthClient AuthClient
SplitFetcher SplitFetcher
SegmentFetcher SegmentFetcher
ImpressionRecorder ImpressionsRecorder
EventRecorder EventsRecorder
MetricRecorder MetricsRecorder
}
// NewSplitAPI creates new splitAPI
func NewSplitAPI(
apikey string,
conf conf.AdvancedConfig,
logger logging.LoggerInterface,
metadata dtos.Metadata,
) *SplitAPI {
return &SplitAPI{
AuthClient: api.NewAuthAPIClient(apikey, conf, logger, metadata),
SplitFetcher: api.NewHTTPSplitFetcher(apikey, conf, logger, metadata),
SegmentFetcher: api.NewHTTPSegmentFetcher(apikey, conf, logger, metadata),
ImpressionRecorder: api.NewHTTPImpressionRecorder(apikey, conf, logger),
EventRecorder: api.NewHTTPEventsRecorder(apikey, conf, logger),
MetricRecorder: api.NewHTTPMetricsRecorder(apikey, conf, logger),
}
}

44
vendor/github.com/splitio/go-split-commons/v3/storage/inmemory/helpers.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,44 @@
package inmemory
import (
"errors"
"fmt"
"sync/atomic"
)
// ErrorOutOfBounds err
var ErrorOutOfBounds error = errors.New("out of bounds")
// AtomicInt64Slice var
type AtomicInt64Slice []int64
// NewAtomicInt64Slice create slice
func NewAtomicInt64Slice(size int64) (AtomicInt64Slice, error) {
if size <= 0 {
return nil, fmt.Errorf("invalid array size: %d", size)
}
return make([]int64, size), nil
}
// Incr increments inx count
func (a AtomicInt64Slice) Incr(index int) {
atomic.AddInt64(&a[index], 1)
}
// FetchAndClearOne returns previous and reset
func (a AtomicInt64Slice) FetchAndClearOne(index int) (int64, error) {
if index >= len(a) || index < 0 {
return 0, ErrorOutOfBounds
}
return atomic.SwapInt64(&a[index], 0), nil
}
// FetchAndClearAll returns all and reset
func (a AtomicInt64Slice) FetchAndClearAll() []int64 {
toRet := make([]int64, len(a))
for index := 0; index < len(a); index++ {
toRet[index] = atomic.SwapInt64(&a[index], 0)
}
return toRet
}

Просмотреть файл

@@ -86,3 +86,14 @@ func (m *MMSegmentStorage) Update(name string, toAdd *set.ThreadUnsafeSet, toRem
// CountRemovedKeys method
func (m *MMSegmentStorage) CountRemovedKeys(segmentName string) int64 { return 0 }
// SegmentKeysCount
func (m *MMSegmentStorage) SegmentKeysCount() int64 {
m.mutex.RLock()
defer m.mutex.RUnlock()
var toReturn int64 = 0
for _, keys := range m.data {
toReturn = toReturn + int64(keys.Size())
}
return toReturn
}

Просмотреть файл

Просмотреть файл

@@ -6,6 +6,8 @@ import (
"sync"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -13,13 +15,14 @@ import (
const MaxAccumulatedBytes = 5 * 1024 * 1024
// NewMQEventsStorage returns an instance of MQEventsStorage
func NewMQEventsStorage(queueSize int, isFull chan string, logger logging.LoggerInterface) *MQEventsStorage {
func NewMQEventsStorage(queueSize int, isFull chan string, logger logging.LoggerInterface, runtimeTelemetry storage.TelemetryRuntimeProducer) *MQEventsStorage {
return &MQEventsStorage{
queue: list.New(),
size: queueSize,
mutexQueue: &sync.Mutex{},
fullChan: isFull,
logger: logger,
queue: list.New(),
size: queueSize,
mutexQueue: &sync.Mutex{},
fullChan: isFull,
logger: logger,
runtimeTelemetry: runtimeTelemetry,
}
}
@@ -36,6 +39,7 @@ type MQEventsStorage struct {
mutexQueue *sync.Mutex
fullChan chan string //only write channel
logger logging.LoggerInterface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
func (s *MQEventsStorage) sendSignalIsFull() {
@@ -55,12 +59,14 @@ func (s *MQEventsStorage) Push(event dtos.EventDTO, size int) error {
defer s.mutexQueue.Unlock()
if s.queue.Len()+1 > s.size {
s.runtimeTelemetry.RecordEventsStats(telemetry.EventsDropped, 1)
s.sendSignalIsFull()
return ErrorMaxSizeReached
}
// Add element
s.queue.PushBack(eventWrapper{event: event, size: size})
s.runtimeTelemetry.RecordEventsStats(telemetry.EventsQueued, 1)
s.accumulatedBytes += size
if s.queue.Len() == s.size || s.accumulatedBytes >= MaxAccumulatedBytes {
s.sendSignalIsFull()

Просмотреть файл

@@ -5,27 +5,31 @@ import (
"sync"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/logging"
)
// NewMQImpressionsStorage returns an instance of MQEventsStorage
func NewMQImpressionsStorage(queueSize int, isFull chan<- string, logger logging.LoggerInterface) *MQImpressionsStorage {
func NewMQImpressionsStorage(queueSize int, isFull chan<- string, logger logging.LoggerInterface, runtimeTelemetry storage.TelemetryRuntimeProducer) *MQImpressionsStorage {
return &MQImpressionsStorage{
queue: list.New(),
size: queueSize,
mutexQueue: &sync.Mutex{},
fullChan: isFull,
logger: logger,
queue: list.New(),
size: queueSize,
mutexQueue: &sync.Mutex{},
fullChan: isFull,
logger: logger,
runtimeTelemetry: runtimeTelemetry,
}
}
// MQImpressionsStorage in memory events storage
type MQImpressionsStorage struct {
queue *list.List
size int
mutexQueue *sync.Mutex
fullChan chan<- string //only write channel
logger logging.LoggerInterface
queue *list.List
size int
mutexQueue *sync.Mutex
fullChan chan<- string //only write channel
logger logging.LoggerInterface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
func (s *MQImpressionsStorage) sendSignalIsFull() {
@@ -59,13 +63,18 @@ func (s *MQImpressionsStorage) LogImpressions(impressions []dtos.Impression) err
s.mutexQueue.Lock()
defer s.mutexQueue.Unlock()
impressionsToAdd := len(impressions)
for _, impression := range impressions {
if s.queue.Len()+1 > s.size {
s.sendSignalIsFull()
s.runtimeTelemetry.RecordImpressionsStats(telemetry.ImpressionsDropped, int64(impressionsToAdd))
return ErrorMaxSizeReached
}
// Add element
s.queue.PushBack(impression)
s.runtimeTelemetry.RecordImpressionsStats(telemetry.ImpressionsQueued, 1)
impressionsToAdd--
if s.queue.Len() == s.size {
s.sendSignalIsFull()

480
vendor/github.com/splitio/go-split-commons/v3/storage/inmemory/telemetry.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,480 @@
package inmemory
import (
"fmt"
"sync"
"sync/atomic"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/storage"
constants "github.com/splitio/go-split-commons/v3/telemetry"
)
type latencies struct {
// MethodLatencies
treatment AtomicInt64Slice
treatments AtomicInt64Slice
treatmentWithConfig AtomicInt64Slice
treatmentsWithConfig AtomicInt64Slice
track AtomicInt64Slice
// HTTPLatencies
splits AtomicInt64Slice
segments AtomicInt64Slice
impressions AtomicInt64Slice
impressionsCount AtomicInt64Slice
events AtomicInt64Slice
telemetry AtomicInt64Slice
token AtomicInt64Slice
}
type counters struct {
// Evaluation Counters
treatment int64
treatments int64
treatmentWithConfig int64
treatmentsWithConfig int64
track int64
// Push Counters
authRejections int64
tokenRefreshes int64
// Factory Counters
burTimeouts int64
nonReadyUsages int64
}
type records struct {
// Impressions Data
impressionsQueued int64
impressionsDropped int64
impressionsDeduped int64
// Events Data
eventsQueued int64
eventsDropped int64
// LastSynchronization
splits int64
segments int64
impressions int64
impressionsCount int64
events int64
token int64
telemetry int64
// SDK
session int64
}
// TelemetryStorage In Memory Telemetry Storage struct
type TelemetryStorage struct {
counters counters
httpErrors dtos.HTTPErrors
mutexHTTPErrors sync.RWMutex
latencies latencies
records records
streamingEvents []dtos.StreamingEvent // Max Length 20
mutexStreamingEvents sync.RWMutex
tags []string
mutexTags sync.RWMutex
}
// NewTelemetryStorage builds in memory telemetry storage
func NewTelemetryStorage() (storage.TelemetryStorage, error) {
treatmentLatencies, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
treatmentWithConfigLatencies, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
treatmentsLatencies, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
treatmentsWithConfigLatencies, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
track, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
splits, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
segments, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
impressions, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
impressionsCount, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
events, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
telemetry, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
token, err := NewAtomicInt64Slice(constants.LatencyBucketCount)
if err != nil {
return nil, fmt.Errorf("could not create InMemory Storage, %w", err)
}
return &TelemetryStorage{
counters: counters{},
httpErrors: dtos.HTTPErrors{
Splits: make(map[int]int64),
Segments: make(map[int]int64),
Impressions: make(map[int]int64),
ImpressionsCount: make(map[int]int64),
Events: make(map[int]int64),
Token: make(map[int]int64),
Telemetry: make(map[int]int64),
},
mutexHTTPErrors: sync.RWMutex{},
latencies: latencies{
treatment: treatmentLatencies,
treatmentWithConfig: treatmentWithConfigLatencies,
treatments: treatmentsLatencies,
treatmentsWithConfig: treatmentsWithConfigLatencies,
track: track,
splits: splits,
segments: segments,
impressions: impressions,
impressionsCount: impressionsCount,
events: events,
token: token,
telemetry: telemetry,
},
records: records{},
streamingEvents: make([]dtos.StreamingEvent, 0, constants.MaxStreamingEvents),
mutexStreamingEvents: sync.RWMutex{},
tags: make([]string, 0, constants.MaxTags),
mutexTags: sync.RWMutex{},
}, nil
}
// TELEMETRY STORAGE PRODUCER
func (i *TelemetryStorage) RecordConfigData(configData dtos.Config) error {
// No-Op. Config Data will be sent directly to Split Servers. No need to store.
return nil
}
// RecordLatency stores latency for method
func (i *TelemetryStorage) RecordLatency(method string, latency int64) {
bucket := constants.Bucket(latency)
switch method {
case constants.Treatment:
i.latencies.treatment.Incr(bucket)
case constants.Treatments:
i.latencies.treatments.Incr(bucket)
case constants.TreatmentWithConfig:
i.latencies.treatmentWithConfig.Incr(bucket)
case constants.TreatmentsWithConfig:
i.latencies.treatmentsWithConfig.Incr(bucket)
case constants.Track:
i.latencies.track.Incr(bucket)
}
}
// RecordException stores exceptions for method
func (i *TelemetryStorage) RecordException(method string) {
switch method {
case constants.Treatment:
atomic.AddInt64(&i.counters.treatment, 1)
case constants.Treatments:
atomic.AddInt64(&i.counters.treatments, 1)
case constants.TreatmentWithConfig:
atomic.AddInt64(&i.counters.treatmentWithConfig, 1)
case constants.TreatmentsWithConfig:
atomic.AddInt64(&i.counters.treatmentsWithConfig, 1)
case constants.Track:
atomic.AddInt64(&i.counters.track, 1)
}
}
// RecordImpressionsStats records impressions by type
func (i *TelemetryStorage) RecordImpressionsStats(dataType int, count int64) {
switch dataType {
case constants.ImpressionsDropped:
atomic.AddInt64(&i.records.impressionsDropped, count)
case constants.ImpressionsDeduped:
atomic.AddInt64(&i.records.impressionsDeduped, count)
case constants.ImpressionsQueued:
atomic.AddInt64(&i.records.impressionsQueued, count)
}
}
// RecordEventsStats recirds events by type
func (i *TelemetryStorage) RecordEventsStats(dataType int, count int64) {
switch dataType {
case constants.EventsDropped:
atomic.AddInt64(&i.records.eventsDropped, count)
case constants.EventsQueued:
atomic.AddInt64(&i.records.eventsQueued, count)
}
}
// RecordSuccessfulSync records sync for resource
func (i *TelemetryStorage) RecordSuccessfulSync(resource int, timestamp int64) {
switch resource {
case constants.SplitSync:
atomic.StoreInt64(&i.records.splits, timestamp)
case constants.SegmentSync:
atomic.StoreInt64(&i.records.segments, timestamp)
case constants.ImpressionSync:
atomic.StoreInt64(&i.records.impressions, timestamp)
case constants.ImpressionCountSync:
atomic.StoreInt64(&i.records.impressionsCount, timestamp)
case constants.EventSync:
atomic.StoreInt64(&i.records.events, timestamp)
case constants.TelemetrySync:
atomic.StoreInt64(&i.records.telemetry, timestamp)
case constants.TokenSync:
atomic.StoreInt64(&i.records.token, timestamp)
}
}
func (i *TelemetryStorage) createOrUpdate(status int, item map[int]int64) {
if item == nil {
item[status] = 1
return
}
item[status]++
}
// RecordSyncError records http error
func (i *TelemetryStorage) RecordSyncError(resource int, status int) {
i.mutexHTTPErrors.Lock()
defer i.mutexHTTPErrors.Unlock()
switch resource {
case constants.SplitSync:
i.createOrUpdate(status, i.httpErrors.Splits)
case constants.SegmentSync:
i.createOrUpdate(status, i.httpErrors.Segments)
case constants.ImpressionSync:
i.createOrUpdate(status, i.httpErrors.Impressions)
case constants.ImpressionCountSync:
i.createOrUpdate(status, i.httpErrors.ImpressionsCount)
case constants.EventSync:
i.createOrUpdate(status, i.httpErrors.Events)
case constants.TelemetrySync:
i.createOrUpdate(status, i.httpErrors.Telemetry)
case constants.TokenSync:
i.createOrUpdate(status, i.httpErrors.Token)
}
}
// RecordSyncLatency records http error
func (i *TelemetryStorage) RecordSyncLatency(resource int, latency int64) {
bucket := constants.Bucket(latency)
switch resource {
case constants.SplitSync:
i.latencies.splits.Incr(bucket)
case constants.SegmentSync:
i.latencies.segments.Incr(bucket)
case constants.ImpressionSync:
i.latencies.impressions.Incr(bucket)
case constants.ImpressionCountSync:
i.latencies.impressionsCount.Incr(bucket)
case constants.EventSync:
i.latencies.events.Incr(bucket)
case constants.TelemetrySync:
i.latencies.telemetry.Incr(bucket)
case constants.TokenSync:
i.latencies.token.Incr(bucket)
}
}
// RecordAuthRejections records auth rejections
func (i *TelemetryStorage) RecordAuthRejections() {
atomic.AddInt64(&i.counters.authRejections, 1)
}
// RecordTokenRefreshes records token
func (i *TelemetryStorage) RecordTokenRefreshes() {
atomic.AddInt64(&i.counters.tokenRefreshes, 1)
}
// RecordStreamingEvent appends new streaming event
func (i *TelemetryStorage) RecordStreamingEvent(event *dtos.StreamingEvent) {
if event == nil {
return
}
i.mutexStreamingEvents.Lock()
defer i.mutexStreamingEvents.Unlock()
if len(i.streamingEvents) < constants.MaxStreamingEvents {
i.streamingEvents = append(i.streamingEvents, *event)
}
}
// AddTag adds particular tag
func (i *TelemetryStorage) AddTag(tag string) {
i.mutexTags.Lock()
defer i.mutexTags.Unlock()
if len(i.tags) < constants.MaxTags {
i.tags = append(i.tags, tag)
}
}
// RecordSessionLength records session length
func (i *TelemetryStorage) RecordSessionLength(session int64) {
atomic.StoreInt64(&i.records.session, session)
}
// RecordNonReadyUsage records non ready usage
func (i *TelemetryStorage) RecordNonReadyUsage() {
atomic.AddInt64(&i.counters.nonReadyUsages, 1)
}
// RecordBURTimeout records bur timeodout
func (i *TelemetryStorage) RecordBURTimeout() {
atomic.AddInt64(&i.counters.burTimeouts, 1)
}
// TELEMETRY STORAGE CONSUMER
// PopLatencies gets and clears method latencies
func (i *TelemetryStorage) PopLatencies() dtos.MethodLatencies {
return dtos.MethodLatencies{
Treatment: i.latencies.treatment.FetchAndClearAll(),
Treatments: i.latencies.treatments.FetchAndClearAll(),
TreatmentWithConfig: i.latencies.treatmentWithConfig.FetchAndClearAll(),
TreatmentsWithConfig: i.latencies.treatmentsWithConfig.FetchAndClearAll(),
Track: i.latencies.track.FetchAndClearAll(),
}
}
// PopExceptions gets and clears method exceptions
func (i *TelemetryStorage) PopExceptions() dtos.MethodExceptions {
return dtos.MethodExceptions{
Treatment: atomic.SwapInt64(&i.counters.treatment, 0),
Treatments: atomic.SwapInt64(&i.counters.treatments, 0),
TreatmentWithConfig: atomic.SwapInt64(&i.counters.treatmentWithConfig, 0),
TreatmentsWithConfig: atomic.SwapInt64(&i.counters.treatmentsWithConfig, 0),
Track: atomic.SwapInt64(&i.counters.track, 0),
}
}
// GetImpressionsStats gets impressions by type
func (i *TelemetryStorage) GetImpressionsStats(dataType int) int64 {
switch dataType {
case constants.ImpressionsDropped:
return atomic.LoadInt64(&i.records.impressionsDropped)
case constants.ImpressionsDeduped:
return atomic.LoadInt64(&i.records.impressionsDeduped)
case constants.ImpressionsQueued:
return atomic.LoadInt64(&i.records.impressionsQueued)
}
return 0
}
// GetEventsStats gets events by type
func (i *TelemetryStorage) GetEventsStats(dataType int) int64 {
switch dataType {
case constants.EventsDropped:
return atomic.LoadInt64(&i.records.eventsDropped)
case constants.EventsQueued:
return atomic.LoadInt64(&i.records.eventsQueued)
}
return 0
}
// GetLastSynchronization gets last synchronization stats for fetchers and recorders
func (i *TelemetryStorage) GetLastSynchronization() dtos.LastSynchronization {
return dtos.LastSynchronization{
Splits: atomic.LoadInt64(&i.records.splits),
Segments: atomic.LoadInt64(&i.records.segments),
Impressions: atomic.LoadInt64(&i.records.impressions),
ImpressionsCount: atomic.LoadInt64(&i.records.impressionsCount),
Events: atomic.LoadInt64(&i.records.events),
Telemetry: atomic.LoadInt64(&i.records.telemetry),
Token: atomic.LoadInt64(&i.records.token),
}
}
// PopHTTPErrors gets http errors
func (i *TelemetryStorage) PopHTTPErrors() dtos.HTTPErrors {
i.mutexHTTPErrors.Lock()
defer i.mutexHTTPErrors.Unlock()
toReturn := i.httpErrors
i.httpErrors.Splits = make(map[int]int64)
i.httpErrors.Segments = make(map[int]int64)
i.httpErrors.Impressions = make(map[int]int64)
i.httpErrors.ImpressionsCount = make(map[int]int64)
i.httpErrors.Events = make(map[int]int64)
i.httpErrors.Telemetry = make(map[int]int64)
i.httpErrors.Token = make(map[int]int64)
return toReturn
}
// PopHTTPLatencies gets http latencies
func (i *TelemetryStorage) PopHTTPLatencies() dtos.HTTPLatencies {
return dtos.HTTPLatencies{
Splits: i.latencies.splits.FetchAndClearAll(),
Segments: i.latencies.segments.FetchAndClearAll(),
Impressions: i.latencies.impressions.FetchAndClearAll(),
ImpressionsCount: i.latencies.impressionsCount.FetchAndClearAll(),
Events: i.latencies.events.FetchAndClearAll(),
Telemetry: i.latencies.telemetry.FetchAndClearAll(),
Token: i.latencies.token.FetchAndClearAll(),
}
}
// PopAuthRejections gets total amount of auth rejections
func (i *TelemetryStorage) PopAuthRejections() int64 {
return atomic.SwapInt64(&i.counters.authRejections, 0)
}
// PopTokenRefreshes gets total amount of token refreshes
func (i *TelemetryStorage) PopTokenRefreshes() int64 {
return atomic.SwapInt64(&i.counters.tokenRefreshes, 0)
}
// PopStreamingEvents gets streamingEvents data
func (i *TelemetryStorage) PopStreamingEvents() []dtos.StreamingEvent {
i.mutexStreamingEvents.Lock()
defer i.mutexStreamingEvents.Unlock()
toReturn := i.streamingEvents
i.streamingEvents = make([]dtos.StreamingEvent, 0, constants.MaxStreamingEvents)
return toReturn
}
// PopTags gets total amount of tags
func (i *TelemetryStorage) PopTags() []string {
i.mutexTags.Lock()
defer i.mutexTags.Unlock()
toReturn := i.tags
i.tags = make([]string, 0, constants.MaxTags)
return toReturn
}
// GetSessionLength gets session duration
func (i *TelemetryStorage) GetSessionLength() int64 {
return atomic.LoadInt64(&i.records.session)
}
// GetNonReadyUsages gets non usages on ready
func (i *TelemetryStorage) GetNonReadyUsages() int64 {
return atomic.LoadInt64(&i.counters.nonReadyUsages)
}
// GetBURTimeouts gets timedouts data
func (i *TelemetryStorage) GetBURTimeouts() int64 {
return atomic.LoadInt64(&i.counters.burTimeouts)
}

105
vendor/github.com/splitio/go-split-commons/v3/storage/interfaces.go сгенерированный поставляемый
Просмотреть файл

@@ -36,6 +36,7 @@ type SegmentStorageConsumer interface {
CountRemovedKeys(segmentName string) int64
Keys(segmentName string) *set.ThreadUnsafeSet
SegmentContainsKey(segmentName string, key string) (bool, error)
SegmentKeysCount() int64
}
// ImpressionStorageProducer interface should be impemented by structs that accept incoming impressions
@@ -52,25 +53,6 @@ type ImpressionStorageConsumer interface {
PopNWithMetadata(n int64) ([]dtos.ImpressionQueueObject, error)
}
// MetricsStorageProducer interface should be impemented by structs that accept incoming metrics
type MetricsStorageProducer interface {
PutGauge(key string, gauge float64)
IncLatency(metricName string, index int)
IncCounter(key string)
}
// MetricsStorageConsumer interface should be implemented by structs that offer popping metrics
type MetricsStorageConsumer interface {
PeekCounters() map[string]int64
PeekLatencies() map[string][]int64
PopGauges() []dtos.GaugeDTO
PopLatencies() []dtos.LatenciesDTO
PopCounters() []dtos.CounterDTO
PopGaugesWithMetadata() (*dtos.GaugeDataBulk, error)
PopLatenciesWithMetadata() (*dtos.LatencyDataBulk, error)
PopCountersWithMetadata() (*dtos.CounterDataBulk, error)
}
// EventStorageProducer interface should be implemented by structs that accept incoming events
type EventStorageProducer interface {
Push(event dtos.EventDTO, size int) error
@@ -85,6 +67,79 @@ type EventStorageConsumer interface {
PopNWithMetadata(n int64) ([]dtos.QueueStoredEventDTO, error)
}
// TelemetryStorageProducer interface should be implemented by struct that accepts incoming telemetry
type TelemetryStorageProducer interface {
TelemetryConfigProducer
TelemetryEvaluationProducer
TelemetryRuntimeProducer
}
// TelemetryRedisProducer interface redis
type TelemetryRedisProducer interface {
TelemetryConfigProducer
TelemetryEvaluationProducer
}
// TelemetryConfigProducer interface for config data
type TelemetryConfigProducer interface {
RecordConfigData(configData dtos.Config) error
RecordNonReadyUsage()
RecordBURTimeout()
}
// TelemetryEvaluationProducer for evaluation
type TelemetryEvaluationProducer interface {
RecordLatency(method string, latency int64)
RecordException(method string)
}
// TelemetryRuntimeProducer for runtime stats
type TelemetryRuntimeProducer interface {
AddTag(tag string)
RecordImpressionsStats(dataType int, count int64)
RecordEventsStats(dataType int, count int64)
RecordSuccessfulSync(resource int, time int64)
RecordSyncError(resource int, status int)
RecordSyncLatency(resource int, latency int64)
RecordAuthRejections()
RecordTokenRefreshes()
RecordStreamingEvent(streamingEvent *dtos.StreamingEvent)
RecordSessionLength(session int64)
}
// TelemetryStorageConsumer interface should be implemented by structs that offer popping telemetry
type TelemetryStorageConsumer interface {
TelemetryConfigConsumer
TelemetryEvaluationConsumer
TelemetryRuntimeConsumer
}
// TelemetryConfigConsumer interface for config data
type TelemetryConfigConsumer interface {
GetNonReadyUsages() int64
GetBURTimeouts() int64
}
// TelemetryEvaluationConsumer for evaluation
type TelemetryEvaluationConsumer interface {
PopLatencies() dtos.MethodLatencies
PopExceptions() dtos.MethodExceptions
}
// TelemetryRuntimeConsumer for runtime stats
type TelemetryRuntimeConsumer interface {
GetImpressionsStats(dataType int) int64
GetEventsStats(dataType int) int64
GetLastSynchronization() dtos.LastSynchronization
PopHTTPErrors() dtos.HTTPErrors
PopHTTPLatencies() dtos.HTTPLatencies
PopAuthRejections() int64
PopTokenRefreshes() int64
PopStreamingEvents() []dtos.StreamingEvent
PopTags() []string
GetSessionLength() int64
}
// --- Wide Interfaces
// SplitStorage wraps consumer & producer interfaces
@@ -105,14 +160,14 @@ type ImpressionStorage interface {
ImpressionStorageProducer
}
// MetricsStorage wraps consumer and producer interfaces
type MetricsStorage interface {
MetricsStorageConsumer
MetricsStorageProducer
}
// EventsStorage wraps consumer and producer interfaces
type EventsStorage interface {
EventStorageConsumer
EventStorageProducer
}
// TelemetryStorage wraps consumer and producer interfaces
type TelemetryStorage interface {
TelemetryStorageConsumer
TelemetryStorageProducer
}

125
vendor/github.com/splitio/go-split-commons/v3/storage/metricWrapper.go сгенерированный поставляемый
Просмотреть файл

@@ -1,125 +0,0 @@
package storage
import (
"errors"
"strings"
"github.com/splitio/go-toolkit/v4/logging"
)
// MetricWrapper struct
type MetricWrapper struct {
Telemetry MetricsStorage
LocalTelemetry MetricsStorage
logger logging.LoggerInterface
}
const (
// SplitChangesCounter counters
SplitChangesCounter = iota
// SplitChangesLatency latencies
SplitChangesLatency
// SegmentChangesCounter counters
SegmentChangesCounter
// SegmentChangesLatency latencies
SegmentChangesLatency
// TestImpressionsCounter counter
TestImpressionsCounter
// TestImpressionsLatency latencies
TestImpressionsLatency
// PostEventsCounter counter
PostEventsCounter
//PostEventsLatency latencies
PostEventsLatency
// MySegmentsCounter counters
MySegmentsCounter
// MySegmentsLatency latencies
MySegmentsLatency
)
const (
counter = "backend::request.{status}"
splitChangesCounter = "splitChangeFetcher.status.{status}"
splitChangesLatency = "splitChangeFetcher.time"
localSplitChangesLatency = "backend::/api/splitChanges"
segmentChangesCounter = "segmentChangeFetcher.status.{status}"
segmentChangesLatency = "segmentChangeFetcher.time"
localSegmentChangesLatency = "backend::/api/segmentChanges"
testImpressionsCounter = "testImpressions.status.{status}"
testImpressionsLatency = "testImpressions.time"
localTestImpressionsLatency = "backend::/api/testImpressions/bulk"
postEventsCounter = "events.status.{status}"
postEventsLatency = "events.time"
localPostEventsLatency = "backend::/api/events/bulk"
mySegmentsCounter = "mySegments.status.{status}"
mySegmentsLatency = "mySegments.time"
localMySegmentsLatency = "backend::/api/mySegments"
)
// NewMetricWrapper builds new wrapper
func NewMetricWrapper(telemetry MetricsStorage, localTelemetry MetricsStorage, logger logging.LoggerInterface) *MetricWrapper {
return &MetricWrapper{
LocalTelemetry: localTelemetry,
logger: logger,
Telemetry: telemetry,
}
}
func (m *MetricWrapper) getKey(key int) (string, string, error) {
switch key {
case SplitChangesCounter:
return splitChangesCounter, counter, nil
case SplitChangesLatency:
return splitChangesLatency, localSplitChangesLatency, nil
case SegmentChangesCounter:
return segmentChangesCounter, counter, nil
case SegmentChangesLatency:
return segmentChangesLatency, localSegmentChangesLatency, nil
case TestImpressionsCounter:
return testImpressionsCounter, counter, nil
case TestImpressionsLatency:
return testImpressionsLatency, localTestImpressionsLatency, nil
case PostEventsCounter:
return postEventsCounter, counter, nil
case PostEventsLatency:
return postEventsLatency, localPostEventsLatency, nil
case MySegmentsCounter:
return mySegmentsCounter, counter, nil
case MySegmentsLatency:
return mySegmentsLatency, localMySegmentsLatency, nil
default:
return "", "", errors.New("Key does not exist")
}
}
// StoreCounters stores counters
func (m *MetricWrapper) StoreCounters(key int, value string) {
common, local, err := m.getKey(key)
if err != nil {
return
}
if m.LocalTelemetry != nil {
m.LocalTelemetry.IncCounter(strings.Replace(local, "{status}", value, 1))
}
if value == "ok" {
value = "200"
}
m.Telemetry.IncCounter(strings.Replace(common, "{status}", value, 1))
}
// StoreLatencies stores counters
func (m *MetricWrapper) StoreLatencies(key int, bucket int) {
common, local, err := m.getKey(key)
if err != nil {
return
}
if m.LocalTelemetry != nil {
m.LocalTelemetry.IncLatency(local, bucket)
}
m.Telemetry.IncLatency(common, bucket)
}

73
vendor/github.com/splitio/go-split-commons/v3/storage/mocks/metric.go сгенерированный поставляемый
Просмотреть файл

@@ -1,73 +0,0 @@
package mocks
import "github.com/splitio/go-split-commons/v3/dtos"
// MockMetricStorage is a mocked implementation of Metric Storage
type MockMetricStorage struct {
IncCounterCall func(key string)
IncLatencyCall func(metricName string, index int)
PutGaugeCall func(key string, gauge float64)
PopGaugesCall func() []dtos.GaugeDTO
PopLatenciesCall func() []dtos.LatenciesDTO
PopCountersCall func() []dtos.CounterDTO
PeekCountersCall func() map[string]int64
PeekLatenciesCall func() map[string][]int64
PopGaugesWithMetadataCall func() (*dtos.GaugeDataBulk, error)
PopCountersWithMetadataCall func() (*dtos.CounterDataBulk, error)
PopLatenciesWithMetadataCall func() (*dtos.LatencyDataBulk, error)
}
// IncCounter mock
func (m MockMetricStorage) IncCounter(key string) {
m.IncCounterCall(key)
}
// IncLatency mock
func (m MockMetricStorage) IncLatency(metricName string, index int) {
m.IncLatencyCall(metricName, index)
}
// PutGauge mock
func (m MockMetricStorage) PutGauge(key string, gauge float64) {
m.PutGaugeCall(key, gauge)
}
// PopGauges mock
func (m MockMetricStorage) PopGauges() []dtos.GaugeDTO {
return m.PopGaugesCall()
}
// PopLatencies mock
func (m MockMetricStorage) PopLatencies() []dtos.LatenciesDTO {
return m.PopLatenciesCall()
}
// PopCounters mock
func (m MockMetricStorage) PopCounters() []dtos.CounterDTO {
return m.PopCountersCall()
}
// PeekCounters mock
func (m MockMetricStorage) PeekCounters() map[string]int64 {
return m.PeekCountersCall()
}
// PeekLatencies mock
func (m MockMetricStorage) PeekLatencies() map[string][]int64 {
return m.PeekLatenciesCall()
}
// PopGaugesWithMetadata mock
func (m MockMetricStorage) PopGaugesWithMetadata() (*dtos.GaugeDataBulk, error) {
return m.PopGaugesWithMetadataCall()
}
// PopCountersWithMetadata mock
func (m MockMetricStorage) PopCountersWithMetadata() (*dtos.CounterDataBulk, error) {
return m.PopCountersWithMetadataCall()
}
// PopLatenciesWithMetadata mock
func (m MockMetricStorage) PopLatenciesWithMetadata() (*dtos.LatencyDataBulk, error) {
return m.PopLatenciesWithMetadataCall()
}

5
vendor/github.com/splitio/go-split-commons/v3/storage/mocks/segment.go сгенерированный поставляемый
Просмотреть файл

@@ -10,6 +10,7 @@ type MockSegmentStorage struct {
SegmentContainsKeyCall func(segmentName string, key string) (bool, error)
SetChangeNumberCall func(segmentName string, till int64) error
CountRemovedKeysCall func(segmentName string) int64
SegmentKeysCountCall func() int64
}
// ChangeNumber mock
@@ -41,3 +42,7 @@ func (m MockSegmentStorage) SetChangeNumber(segmentName string, till int64) erro
func (m MockSegmentStorage) CountRemovedKeys(segmentName string) int64 {
return m.CountRemovedKeysCall(segmentName)
}
func (m MockSegmentStorage) SegmentKeysCount() int64 {
return m.SegmentKeysCountCall()
}

179
vendor/github.com/splitio/go-split-commons/v3/storage/mocks/telemetry.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,179 @@
package mocks
import "github.com/splitio/go-split-commons/v3/dtos"
// MockTelemetryStorage is a mocked implementation of Telemetry Storage
type MockTelemetryStorage struct {
RecordConfigDataCall func(configData dtos.Config) error
RecordLatencyCall func(method string, latency int64)
RecordExceptionCall func(method string)
RecordImpressionsStatsCall func(dataType int, count int64)
RecordEventsStatsCall func(dataType int, count int64)
RecordSuccessfulSyncCall func(resource int, time int64)
RecordSyncErrorCall func(resource int, status int)
RecordSyncLatencyCall func(resource int, latency int64)
RecordAuthRejectionsCall func()
RecordTokenRefreshesCall func()
RecordStreamingEventCall func(streamingEvent *dtos.StreamingEvent)
AddTagCall func(tag string)
RecordSessionLengthCall func(session int64)
RecordNonReadyUsageCall func()
RecordBURTimeoutCall func()
PopLatenciesCall func() dtos.MethodLatencies
PopExceptionsCall func() dtos.MethodExceptions
GetImpressionsStatsCall func(dataType int) int64
GetEventsStatsCall func(dataType int) int64
GetLastSynchronizationCall func() dtos.LastSynchronization
PopHTTPErrorsCall func() dtos.HTTPErrors
PopHTTPLatenciesCall func() dtos.HTTPLatencies
PopAuthRejectionsCall func() int64
PopTokenRefreshesCall func() int64
PopStreamingEventsCall func() []dtos.StreamingEvent
PopTagsCall func() []string
GetSessionLengthCall func() int64
GetNonReadyUsagesCall func() int64
GetBURTimeoutsCall func() int64
}
// RecordConfig mock
func (m MockTelemetryStorage) RecordConfigData(configData dtos.Config) error {
return m.RecordConfigDataCall(configData)
}
// RecordLatency mock
func (m MockTelemetryStorage) RecordLatency(method string, latency int64) {
m.RecordLatencyCall(method, latency)
}
// RecordException mock
func (m MockTelemetryStorage) RecordException(method string) { m.RecordExceptionCall(method) }
// RecordImpressionsStats mock
func (m MockTelemetryStorage) RecordImpressionsStats(dataType int, count int64) {
m.RecordImpressionsStatsCall(dataType, count)
}
// RecordEventsStats mock
func (m MockTelemetryStorage) RecordEventsStats(dataType int, count int64) {
m.RecordEventsStatsCall(dataType, count)
}
// RecordSuccessfulSync mock
func (m MockTelemetryStorage) RecordSuccessfulSync(resource int, time int64) {
m.RecordSuccessfulSyncCall(resource, time)
}
// RecordSyncError mock
func (m MockTelemetryStorage) RecordSyncError(resource int, status int) {
m.RecordSyncErrorCall(resource, status)
}
// RecordSyncLatency mock
func (m MockTelemetryStorage) RecordSyncLatency(resource int, latency int64) {
m.RecordSyncLatencyCall(resource, latency)
}
// RecordAuthRejections mock
func (m MockTelemetryStorage) RecordAuthRejections() {
m.RecordAuthRejectionsCall()
}
// RecordTokenRefreshes mock
func (m MockTelemetryStorage) RecordTokenRefreshes() {
m.RecordTokenRefreshesCall()
}
// RecordStreamingEvent mock
func (m MockTelemetryStorage) RecordStreamingEvent(streamingEvent *dtos.StreamingEvent) {
m.RecordStreamingEventCall(streamingEvent)
}
// AddTag mock
func (m MockTelemetryStorage) AddTag(tag string) {
m.AddTagCall(tag)
}
// RecordSessionLength mock
func (m MockTelemetryStorage) RecordSessionLength(session int64) {
m.RecordSessionLengthCall(session)
}
// RecordNonReadyUsage mock
func (m MockTelemetryStorage) RecordNonReadyUsage() {
m.RecordNonReadyUsageCall()
}
// RecordBURTimeout mock
func (m MockTelemetryStorage) RecordBURTimeout() {
m.RecordBURTimeoutCall()
}
// PopLatencies mock
func (m MockTelemetryStorage) PopLatencies() dtos.MethodLatencies {
return m.PopLatenciesCall()
}
//PopExceptions mock
func (m MockTelemetryStorage) PopExceptions() dtos.MethodExceptions {
return m.PopExceptionsCall()
}
// GetImpressionsStats mock
func (m MockTelemetryStorage) GetImpressionsStats(dataType int) int64 {
return m.GetImpressionsStatsCall(dataType)
}
// GetEventsStats mock
func (m MockTelemetryStorage) GetEventsStats(dataType int) int64 {
return m.GetEventsStatsCall(dataType)
}
// GetLastSynchronization mock
func (m MockTelemetryStorage) GetLastSynchronization() dtos.LastSynchronization {
return m.GetLastSynchronizationCall()
}
// PopHTTPErrors mock
func (m MockTelemetryStorage) PopHTTPErrors() dtos.HTTPErrors {
return m.PopHTTPErrorsCall()
}
// PopHTTPLatencies mock
func (m MockTelemetryStorage) PopHTTPLatencies() dtos.HTTPLatencies {
return m.PopHTTPLatenciesCall()
}
// PopAuthRejections mock
func (m MockTelemetryStorage) PopAuthRejections() int64 {
return m.PopAuthRejectionsCall()
}
// PopTokenRefreshes mock
func (m MockTelemetryStorage) PopTokenRefreshes() int64 {
return m.PopTokenRefreshesCall()
}
// PopStreamingEvents mock
func (m MockTelemetryStorage) PopStreamingEvents() []dtos.StreamingEvent {
return m.PopStreamingEventsCall()
}
// PopTags mock
func (m MockTelemetryStorage) PopTags() []string {
return m.PopTagsCall()
}
// GetSessionLength mock
func (m MockTelemetryStorage) GetSessionLength() int64 {
return m.GetSessionLengthCall()
}
// GetNonReadyUsages mock
func (m MockTelemetryStorage) GetNonReadyUsages() int64 {
return m.GetNonReadyUsagesCall()
}
// GetBURTimeouts mock
func (m MockTelemetryStorage) GetBURTimeouts() int64 {
return m.GetBURTimeoutsCall()
}

149
vendor/github.com/splitio/go-split-commons/v3/storage/mutexmap/metrics.go сгенерированный поставляемый
Просмотреть файл

@@ -1,149 +0,0 @@
package mutexmap
import (
"sync"
"github.com/splitio/go-split-commons/v3/dtos"
)
// MMMetricsStorage contains an in-memory implementation of Metrics storage
type MMMetricsStorage struct {
gaugeData map[string]float64
gaugeMutex *sync.Mutex
counterData map[string]int64
countersMutex *sync.RWMutex
latenciesData map[string][]int64
latenciesMutex *sync.RWMutex
}
// NewMMMetricsStorage instantiates a new MMMetricsStorage
func NewMMMetricsStorage() *MMMetricsStorage {
return &MMMetricsStorage{
counterData: make(map[string]int64),
countersMutex: &sync.RWMutex{},
gaugeData: make(map[string]float64),
gaugeMutex: &sync.Mutex{},
latenciesData: make(map[string][]int64),
latenciesMutex: &sync.RWMutex{},
}
}
// PutGauge stores a new gauge value for a specific key
func (m *MMMetricsStorage) PutGauge(key string, gauge float64) {
m.gaugeMutex.Lock()
defer m.gaugeMutex.Unlock()
m.gaugeData[key] = gauge
}
// PopGauges returns and deletes all gauges currently stored
func (m *MMMetricsStorage) PopGauges() []dtos.GaugeDTO {
m.gaugeMutex.Lock()
defer func() {
m.gaugeData = make(map[string]float64)
m.gaugeMutex.Unlock()
}()
gauges := make([]dtos.GaugeDTO, 0)
for key, gauge := range m.gaugeData {
gauges = append(gauges, dtos.GaugeDTO{
MetricName: key,
Gauge: gauge,
})
}
return gauges
}
// IncCounter increments the counter for a specific key. It initializes it in 1 if it doesn't exist when this function
// is called.
func (m *MMMetricsStorage) IncCounter(key string) {
m.countersMutex.Lock()
defer m.countersMutex.Unlock()
_, exists := m.counterData[key]
if !exists {
m.counterData[key] = 1
} else {
m.counterData[key]++
}
}
// PopCounters returns and deletes all the counters stored
func (m *MMMetricsStorage) PopCounters() []dtos.CounterDTO {
m.countersMutex.Lock()
defer func() {
m.counterData = make(map[string]int64)
m.countersMutex.Unlock()
}()
counters := make([]dtos.CounterDTO, 0)
for key, counter := range m.counterData {
counters = append(counters, dtos.CounterDTO{
MetricName: key,
Count: counter,
})
}
return counters
}
// PeekCounters returns Counters
func (m *MMMetricsStorage) PeekCounters() map[string]int64 {
m.countersMutex.RLock()
defer m.countersMutex.RUnlock()
return m.counterData
}
// PeekLatencies returns Latencies
func (m *MMMetricsStorage) PeekLatencies() map[string][]int64 {
m.latenciesMutex.RLock()
defer m.latenciesMutex.RUnlock()
return m.latenciesData
}
// IncLatency increments the latency for a specific key and bucket. If the key doesn't exist it's initialized to
// an empty array of 23 items.
func (m *MMMetricsStorage) IncLatency(metricName string, index int) {
if index < 0 || index > 22 {
return
}
m.latenciesMutex.Lock()
defer m.latenciesMutex.Unlock()
_, exists := m.latenciesData[metricName]
if !exists {
m.latenciesData[metricName] = make([]int64, 23)
m.latenciesData[metricName][index] = 1
} else {
m.latenciesData[metricName][index]++
}
}
// PopLatencies Returns and delete all the latencies currently stored
func (m *MMMetricsStorage) PopLatencies() []dtos.LatenciesDTO {
m.latenciesMutex.Lock()
defer func() {
m.latenciesData = make(map[string][]int64)
m.latenciesMutex.Unlock()
}()
latencies := make([]dtos.LatenciesDTO, 0)
for key, latency := range m.latenciesData {
latencies = append(latencies, dtos.LatenciesDTO{
Latencies: latency,
MetricName: key,
})
}
return latencies
}
// PopGaugesWithMetadata mock
func (m *MMMetricsStorage) PopGaugesWithMetadata() (*dtos.GaugeDataBulk, error) {
panic("Not implemented for inmemory")
}
// PopLatenciesWithMetadata mock
func (m *MMMetricsStorage) PopLatenciesWithMetadata() (*dtos.LatencyDataBulk, error) {
panic("Not implemented for inmemory")
}
// PopCountersWithMetadata mock
func (m *MMMetricsStorage) PopCountersWithMetadata() (*dtos.CounterDataBulk, error) {
panic("Not implemented for inmemory")
}

28
vendor/github.com/splitio/go-split-commons/v3/storage/redis/constants.go сгенерированный поставляемый
Просмотреть файл

@@ -1,17 +1,19 @@
package redis
const (
redisSplit = "SPLITIO.split.{split}" // split object
redisSplitTill = "SPLITIO.splits.till" // last split fetch
redisSegment = "SPLITIO.segment.{segment}" // segment object
redisSegmentTill = "SPLITIO.segment.{segment}.till" // last segment fetch
redisImpressions = "SPLITIO/{sdkVersion}/{instanceId}/impressions.{feature}" // impressions for a feature
redisLatency = "SPLITIO/{sdkVersion}/{instanceId}/latency.{metric}.bucket.{bucket}" // latency bucket
redisCounter = "SPLITIO/{sdkVersion}/{instanceId}/count.{metric}" // counter
redisGauge = "SPLITIO/{sdkVersion}/{instanceId}/gauge.{metric}" // gauge
redisEvents = "SPLITIO.events" // events LIST key
redisImpressionsQueue = "SPLITIO.impressions" // impressions LIST key
redisImpressionsTTL = 60 // impressions default TTL
redisTrafficType = "SPLITIO.trafficType.{trafficType}" // traffic Type fetch
redisHash = "SPLITIO.hash"
redisSplit = "SPLITIO.split.{split}" // split object
redisSplitTill = "SPLITIO.splits.till" // last split fetch
redisSegment = "SPLITIO.segment.{segment}" // segment object
redisSegmentTill = "SPLITIO.segment.{segment}.till" // last segment fetch
redisEvents = "SPLITIO.events" // events LIST key
redisImpressionsQueue = "SPLITIO.impressions" // impressions LIST key
redisImpressionsTTL = 3600 // impressions default TTL
redisTrafficType = "SPLITIO.trafficType.{trafficType}" // traffic Type fetch
redisHash = "SPLITIO.hash" // hash key
redisConfig = "SPLITIO.telemetry.config" // config Key
redisConfigTTL = 3600 // config TTL
redisLatency = "SPLITIO.telemetry.latencies" // latency Key
redisExceptionField = "{sdkVersion}/{machineName}/{machineIP}/{method}" // exception field template
redisException = "SPLITIO.telemetry.exceptions" // exception Key
redisLatencyField = "{sdkVersion}/{machineName}/{machineIP}/{method}/{bucket}" // latency field template
)

24
vendor/github.com/splitio/go-split-commons/v3/storage/redis/impressions.go сгенерированный поставляемый
Просмотреть файл

@@ -14,23 +14,21 @@ const impressionsTTLRefresh = time.Duration(3600) * time.Second
// ImpressionStorage is a redis-based implementation of split storage
type ImpressionStorage struct {
client *redis.PrefixedRedisClient
mutex *sync.Mutex
logger logging.LoggerInterface
redisKey string
impressionsTTL time.Duration
metadata dtos.Metadata
client *redis.PrefixedRedisClient
mutex *sync.Mutex
logger logging.LoggerInterface
redisKey string
metadata dtos.Metadata
}
// NewImpressionStorage creates a new RedisSplitStorage and returns a reference to it
func NewImpressionStorage(client *redis.PrefixedRedisClient, metadata dtos.Metadata, logger logging.LoggerInterface) *ImpressionStorage {
return &ImpressionStorage{
client: client,
mutex: &sync.Mutex{},
logger: logger,
redisKey: redisImpressionsQueue,
impressionsTTL: redisImpressionsTTL,
metadata: metadata,
client: client,
mutex: &sync.Mutex{},
logger: logger,
redisKey: redisImpressionsQueue,
metadata: metadata,
}
}
@@ -83,7 +81,7 @@ func (r *ImpressionStorage) push(impressions []dtos.ImpressionQueueObject) error
// Checks if expiration needs to be set
if inserted == int64(len(impressionsJSON)) {
r.logger.Debug("Proceeding to set expiration for: ", r.redisKey)
result := r.client.Expire(r.redisKey, time.Duration(r.impressionsTTL)*time.Minute)
result := r.client.Expire(r.redisKey, time.Duration(redisImpressionsTTL)*time.Second)
if result == false {
r.logger.Error("Something were wrong setting expiration", errPush)
}

322
vendor/github.com/splitio/go-split-commons/v3/storage/redis/metrics.go сгенерированный поставляемый
Просмотреть файл

@@ -1,322 +0,0 @@
package redis
import (
"fmt"
"regexp"
"strconv"
"strings"
"sync"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-toolkit/v4/logging"
"github.com/splitio/go-toolkit/v4/redis"
)
// MetricsStorage is a redis-based implementation of split storage
type MetricsStorage struct {
client redis.PrefixedRedisClient
logger logging.LoggerInterface
gaugeSingleTemplate string
countersSingleTemplate string
latenciesSingleTemplate string
gaugeMultiTemplate string
countersMultiTemplate string
latenciesMultiTemplate string
mutex *sync.RWMutex
}
// NewMetricsStorage creates a new RedisSplitStorage and returns a reference to it
func NewMetricsStorage(redisClient *redis.PrefixedRedisClient, metadata dtos.Metadata, logger logging.LoggerInterface) *MetricsStorage {
// @Todo Split Storages between Go-Client and Redis
gaugeSingleTemplate := strings.Replace(redisGauge, "{sdkVersion}", metadata.SDKVersion, 1)
gaugeSingleTemplate = strings.Replace(gaugeSingleTemplate, "{instanceId}", metadata.MachineName, 1)
countersSingleTemplate := strings.Replace(redisCounter, "{sdkVersion}", metadata.SDKVersion, 1)
countersSingleTemplate = strings.Replace(countersSingleTemplate, "{instanceId}", metadata.MachineName, 1)
latenciesSingleTemplate := strings.Replace(redisLatency, "{sdkVersion}", metadata.SDKVersion, 1)
latenciesSingleTemplate = strings.Replace(latenciesSingleTemplate, "{instanceId}", metadata.MachineName, 1)
gaugeMultiTemplate := strings.Replace(redisGauge, "{sdkVersion}", "*", 1)
gaugeMultiTemplate = strings.Replace(gaugeMultiTemplate, "{instanceId}", "*", 1)
gaugeMultiTemplate = strings.Replace(gaugeMultiTemplate, "{metric}", "*", 1)
countersMultiTemplate := strings.Replace(redisCounter, "{sdkVersion}", "*", 1)
countersMultiTemplate = strings.Replace(countersMultiTemplate, "{instanceId}", "*", 1)
countersMultiTemplate = strings.Replace(countersMultiTemplate, "{metric}", "*", 1)
latenciesMultiTemplate := strings.Replace(redisLatency, "{sdkVersion}", "*", 1)
latenciesMultiTemplate = strings.Replace(latenciesMultiTemplate, "{instanceId}", "*", 1)
latenciesMultiTemplate = strings.Replace(latenciesMultiTemplate, "{metric}", "*", 1)
latenciesMultiTemplate = strings.Replace(latenciesMultiTemplate, "{bucket}", "*", 1)
return &MetricsStorage{
client: *redisClient,
logger: logger,
gaugeSingleTemplate: gaugeSingleTemplate,
countersSingleTemplate: countersSingleTemplate,
latenciesSingleTemplate: latenciesSingleTemplate,
gaugeMultiTemplate: gaugeMultiTemplate,
countersMultiTemplate: countersMultiTemplate,
latenciesMultiTemplate: latenciesMultiTemplate,
mutex: &sync.RWMutex{},
}
}
// IncCounter incraeses the count for a specific metric
func (r *MetricsStorage) IncCounter(metric string) {
keyToIncr := strings.Replace(r.countersSingleTemplate, "{metric}", metric, 1)
_, err := r.client.Incr(keyToIncr)
if err != nil {
r.logger.Error(fmt.Sprintf("Error incrementing counterfor metric \"%s\" in redis: %s", metric, err.Error()))
}
}
// IncLatency incraeses the latency of a bucket for a specific metric
func (r *MetricsStorage) IncLatency(metric string, index int) {
keyToIncr := strings.Replace(r.latenciesSingleTemplate, "{metric}", metric, 1)
keyToIncr = strings.Replace(keyToIncr, "{bucket}", strconv.FormatInt(int64(index), 10), 1)
_, err := r.client.Incr(keyToIncr)
if err != nil {
r.logger.Error(fmt.Sprintf(
"Error incrementing latency bucket %d for metric \"%s\" in redis: %s", index, metric, err.Error(),
))
}
}
// PutGauge stores a gauge in redis
func (r *MetricsStorage) PutGauge(key string, gauge float64) {
keyToStore := strings.Replace(r.gaugeSingleTemplate, "{metric}", key, 1)
err := r.client.Set(keyToStore, gauge, 0)
if err != nil {
r.logger.Error(fmt.Sprintf("Error storing gauge \"%s\" in redis: %s\n", key, err))
}
}
// PopCounters some
func (r *MetricsStorage) PopCounters() []dtos.CounterDTO {
panic("Not implemented for redis")
}
// PopGauges some
func (r *MetricsStorage) PopGauges() []dtos.GaugeDTO {
panic("Not implemented for redis")
}
// PopLatencies some
func (r *MetricsStorage) PopLatencies() []dtos.LatenciesDTO {
panic("Not implemented for redis")
}
func (r *MetricsStorage) popByPattern(pattern string, useTransaction bool) (map[string]interface{}, error) {
r.mutex.Lock()
defer r.mutex.Unlock()
keys, err := r.client.Keys(pattern)
if err != nil {
r.logger.Error(err.Error())
return nil, err
}
if len(keys) == 0 {
return map[string]interface{}{}, nil
}
values, err := r.client.MGet(keys)
if err != nil {
r.logger.Error(err.Error())
return nil, err
}
_, err = r.client.Del(keys...)
if err != nil {
// if we failed to delete the keys, log an error and continue working.
r.logger.Error(err.Error())
}
toReturn := make(map[string]interface{})
for index := range keys {
if index >= len(keys) || index >= len(values) {
break
}
toReturn[keys[index]] = values[index]
}
return toReturn, nil
}
func parseIntRedisValue(s interface{}) (int64, error) {
asStr, ok := s.(string)
if !ok {
return 0, fmt.Errorf("%+v is not a string", s)
}
asInt64, err := strconv.ParseInt(asStr, 10, 64)
if err != nil {
return 0, err
}
return asInt64, nil
}
func parseFloatRedisValue(s interface{}) (float64, error) {
asStr, ok := s.(string)
if !ok {
return 0, fmt.Errorf("%+v is not a string", s)
}
asFloat64, err := strconv.ParseFloat(asStr, 64)
if err != nil {
return 0, err
}
return asFloat64, nil
}
func (r *MetricsStorage) parseLatencyKey(key string) (string, string, string, int, error) {
re := regexp.MustCompile(`(\w+.)?SPLITIO\/([^\/]+)\/([^\/]+)\/latency.([^\/]+).bucket.([0-9]*)`)
match := re.FindStringSubmatch(key)
if len(match) < 6 {
return "", "", "", 0, fmt.Errorf("Error parsing key %s", key)
}
sdkNameAndVersion := match[2]
if sdkNameAndVersion == "" {
return "", "", "", 0, fmt.Errorf("Invalid sdk name/version")
}
machineIP := match[3]
if machineIP == "" {
return "", "", "", 0, fmt.Errorf("Invalid machine IP")
}
metricName := match[4]
if metricName == "" {
return "", "", "", 0, fmt.Errorf("Invalid feature name")
}
bucketNumber, err := strconv.Atoi(match[5])
if err != nil {
return "", "", "", 0, fmt.Errorf("Error parsing bucket number: %s", err.Error())
}
r.logger.Verbose("Impression parsed key", match)
return sdkNameAndVersion, machineIP, metricName, bucketNumber, nil
}
func (r *MetricsStorage) parseMetricKey(metricType string, key string) (string, string, string, error) {
var re = regexp.MustCompile(strings.Replace(
`(\w+.)?SPLITIO\/([^\/]+)\/([^\/]+)\/{metricType}.([\s\S]*)`,
"{metricType}",
metricType,
1,
))
match := re.FindStringSubmatch(key)
if len(match) < 5 {
return "", "", "", fmt.Errorf("Error parsing key %s", key)
}
sdkNameAndVersion := match[2]
if sdkNameAndVersion == "" {
return "", "", "", fmt.Errorf("Invalid sdk name/version")
}
machineIP := match[3]
if machineIP == "" {
return "", "", "", fmt.Errorf("Invalid machine IP")
}
metricName := match[4]
if metricName == "" {
return "", "", "", fmt.Errorf("Invalid feature name")
}
r.logger.Verbose("Impression parsed key", match)
return sdkNameAndVersion, machineIP, metricName, nil
}
// PopGaugesWithMetadata returns gauges values saved in Redis by SDKs
func (r *MetricsStorage) PopGaugesWithMetadata() (*dtos.GaugeDataBulk, error) {
data, err := r.popByPattern(r.gaugeMultiTemplate, false)
if err != nil {
r.logger.Error(err.Error())
return nil, err
}
gaugesToReturn := dtos.NewGaugeDataBulk()
for key, value := range data {
sdkNameAndVersion, machineIP, metricName, err := r.parseMetricKey("gauge", key)
if err != nil {
r.logger.Error(fmt.Sprintf("Unable to parse key %s. Skipping", key))
continue
}
asFloat, err := parseFloatRedisValue(value)
if err != nil {
r.logger.Error(fmt.Sprintf("Unable to parse value %+v. Skipping", value))
continue
}
gaugesToReturn.PutGauge(sdkNameAndVersion, machineIP, metricName, asFloat)
}
return gaugesToReturn, nil
}
// PopCountersWithMetadata returns counter values saved in Redis by SDKs
func (r *MetricsStorage) PopCountersWithMetadata() (*dtos.CounterDataBulk, error) {
data, err := r.popByPattern(r.countersMultiTemplate, false)
if err != nil {
r.logger.Error(err.Error())
return nil, err
}
countersToReturn := dtos.NewCounterDataBulk()
for key, value := range data {
sdkNameAndVersion, machineIP, metricName, err := r.parseMetricKey("count", key)
if err != nil {
r.logger.Error("Unable to parse key %s. Skipping", key)
continue
}
asInt, err := parseIntRedisValue(value)
if err != nil {
r.logger.Error(err.Error())
continue
}
countersToReturn.PutCounter(sdkNameAndVersion, machineIP, metricName, asInt)
}
return countersToReturn, nil
}
// PopLatenciesWithMetadata returns latency values saved in Redis by SDKs
func (r *MetricsStorage) PopLatenciesWithMetadata() (*dtos.LatencyDataBulk, error) {
data, err := r.popByPattern(r.latenciesMultiTemplate, false)
if err != nil {
r.logger.Error(err.Error())
return nil, err
}
latenciesToReturn := dtos.NewLatencyDataBulk()
for key, value := range data {
value, err := parseIntRedisValue(value)
if err != nil {
r.logger.Warning(fmt.Sprintf("Unable to parse value of key %s. Skipping", key))
continue
}
sdkNameAndVersion, machineIP, metricName, bucketNumber, err := r.parseLatencyKey(key)
if err != nil {
r.logger.Warning(fmt.Sprintf("Unable to parse key %s. Skipping", key))
continue
}
latenciesToReturn.PutLatency(sdkNameAndVersion, machineIP, metricName, bucketNumber, value)
}
r.logger.Verbose(latenciesToReturn)
return latenciesToReturn, nil
}
// PeekCounters returns Counters
func (r *MetricsStorage) PeekCounters() map[string]int64 {
return make(map[string]int64, 0)
}
// PeekLatencies returns Latencies
func (r *MetricsStorage) PeekLatencies() map[string][]int64 {
return make(map[string][]int64, 0)
}

6
vendor/github.com/splitio/go-split-commons/v3/storage/redis/segments.go сгенерированный поставляемый
Просмотреть файл

@@ -6,6 +6,7 @@ import (
"strings"
"sync"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-toolkit/v4/datastructures/set"
"github.com/splitio/go-toolkit/v4/logging"
"github.com/splitio/go-toolkit/v4/redis"
@@ -19,7 +20,7 @@ type SegmentStorage struct {
}
// NewSegmentStorage creates a new RedisSegmentStorage and returns a reference to it
func NewSegmentStorage(redisClient *redis.PrefixedRedisClient, logger logging.LoggerInterface) *SegmentStorage {
func NewSegmentStorage(redisClient *redis.PrefixedRedisClient, logger logging.LoggerInterface) storage.SegmentStorage {
return &SegmentStorage{
client: *redisClient,
logger: logger,
@@ -98,3 +99,6 @@ func (r *SegmentStorage) SegmentContainsKey(segmentName string, key string) (boo
// CountRemovedKeys method
func (r *SegmentStorage) CountRemovedKeys(segmentName string) int64 { return 0 }
// SegmentKeysCount method
func (r *SegmentStorage) SegmentKeysCount() int64 { return 0 }

105
vendor/github.com/splitio/go-split-commons/v3/storage/redis/telemetry.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,105 @@
package redis
import (
"encoding/json"
"fmt"
"strings"
"time"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/logging"
"github.com/splitio/go-toolkit/v4/redis"
)
const (
sdkVersion = "{sdkVersion}"
machineIP = "{machineIP}"
machineName = "{machineName}"
name = "{method}"
bucketName = "{bucket}"
)
// TelemetryStorage is a redis-based implementation of telemetry storage
type TelemetryStorage struct {
client *redis.PrefixedRedisClient
exceptionTemplate string
latencyTemplate string
logger logging.LoggerInterface
metadata dtos.Metadata
}
// NewTelemetryStorage creates a new RedisTelemetryStorage and returns a reference to it
func NewTelemetryStorage(redisClient *redis.PrefixedRedisClient, logger logging.LoggerInterface, metadata dtos.Metadata) storage.TelemetryRedisProducer {
replacer := strings.NewReplacer(sdkVersion, metadata.SDKVersion, machineName, metadata.MachineName, machineIP, metadata.MachineIP)
exceptionTemplate := replacer.Replace(redisExceptionField)
latencyTemplate := replacer.Replace(redisLatencyField)
return &TelemetryStorage{
client: redisClient,
exceptionTemplate: exceptionTemplate,
latencyTemplate: latencyTemplate,
logger: logger,
metadata: metadata,
}
}
// TELEMETRY STORAGE PRODUCER
// RecordConfigData push config into queue
func (t *TelemetryStorage) RecordConfigData(configData dtos.Config) error {
jsonData, err := json.Marshal(dtos.TelemetryQueueObject{
Metadata: t.metadata,
Config: configData,
})
if err != nil {
t.logger.Error("Error encoding impression in json", err.Error())
}
inserted, errPush := t.client.RPush(redisConfig, jsonData)
if errPush != nil {
t.logger.Error("Something were wrong pushing config data to redis", errPush)
return errPush
}
// Checks if expiration needs to be set
if inserted == 1 {
t.logger.Debug("Proceeding to set expiration for: ", redisConfig)
result := t.client.Expire(redisConfig, time.Duration(redisConfigTTL)*time.Second)
if !result {
t.logger.Error("Something were wrong setting expiration", errPush)
}
}
return nil
}
// RecordLatency stores latency for method
func (t *TelemetryStorage) RecordLatency(method string, latency int64) {
bucket := telemetry.Bucket(latency)
field := strings.Replace(t.latencyTemplate, name, method, 1)
field = strings.Replace(field, bucketName, fmt.Sprintf("%d", bucket), 1)
_, err := t.client.HIncrBy(redisLatency, field, 1)
if err != nil {
t.logger.Error("Error recording in redis.", err.Error())
}
}
// RecordException stores exceptions for method
func (t *TelemetryStorage) RecordException(method string) {
field := strings.Replace(t.exceptionTemplate, name, method, 1)
_, err := t.client.HIncrBy(redisException, field, 1)
if err != nil {
t.logger.Error("Error recording in redis.", err.Error())
}
}
// RecordNonReadyUsage records non ready usage
func (t *TelemetryStorage) RecordNonReadyUsage() {
// No-Op. Redis is implicitly ready and does not need to wait for anything. Tracking not required.
}
// RecordBURTimeout records bur timeodout
func (t *TelemetryStorage) RecordBURTimeout() {
// No-Op. Redis is implicitly ready and does not need to block for anything. Tracking not required.
}

29
vendor/github.com/splitio/go-split-commons/v3/synchronizer/local.go сгенерированный поставляемый
Просмотреть файл

@@ -1,10 +1,8 @@
package synchronizer
import (
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/service/api"
"github.com/splitio/go-split-commons/v3/storage"
storageMock "github.com/splitio/go-split-commons/v3/storage/mocks"
"github.com/splitio/go-split-commons/v3/synchronizer/worker/split"
"github.com/splitio/go-split-commons/v3/tasks"
"github.com/splitio/go-toolkit/v4/logging"
@@ -12,30 +10,15 @@ import (
// Local implements Local Synchronizer
type Local struct {
splitTasks SplitTasks
workers Workers
logger logging.LoggerInterface
inMememoryFullQueue chan string
splitTasks SplitTasks
workers Workers
logger logging.LoggerInterface
}
// NewLocal creates new Local
func NewLocal(
period int,
splitAPI *service.SplitAPI,
splitStorage storage.SplitStorage,
logger logging.LoggerInterface,
) Synchronizer {
metricStorageMock := storageMock.MockMetricStorage{
IncCounterCall: func(key string) {},
IncLatencyCall: func(metricName string, index int) {},
PopCountersCall: func() []dtos.CounterDTO { return make([]dtos.CounterDTO, 0, 0) },
PopGaugesCall: func() []dtos.GaugeDTO { return make([]dtos.GaugeDTO, 0, 0) },
PopLatenciesCall: func() []dtos.LatenciesDTO { return make([]dtos.LatenciesDTO, 0, 0) },
PutGaugeCall: func(key string, gauge float64) {},
}
metricsWrapper := storage.NewMetricWrapper(metricStorageMock, nil, logger)
func NewLocal(period int, splitAPI *api.SplitAPI, splitStorage storage.SplitStorage, logger logging.LoggerInterface, runtimeTelemetry storage.TelemetryRuntimeProducer) Synchronizer {
workers := Workers{
SplitFetcher: split.NewSplitFetcher(splitStorage, splitAPI.SplitFetcher, metricsWrapper, logger),
SplitFetcher: split.NewSplitFetcher(splitStorage, splitAPI.SplitFetcher, logger, runtimeTelemetry),
}
return &Local{
splitTasks: SplitTasks{

50
vendor/github.com/splitio/go-split-commons/v3/synchronizer/manager.go сгенерированный поставляемый
Просмотреть файл

@@ -6,9 +6,11 @@ import (
"time"
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/push"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/backoff"
"github.com/splitio/go-toolkit/v4/logging"
"github.com/splitio/go-toolkit/v4/struct/traits/lifecycle"
@@ -38,15 +40,16 @@ type Manager interface {
// ManagerImpl struct
type ManagerImpl struct {
synchronizer Synchronizer
logger logging.LoggerInterface
config conf.AdvancedConfig
pushManager push.Manager
managerStatus chan int
streamingStatus chan int64
operationMode int32
lifecycle lifecycle.Manager
backoff backoff.Interface
synchronizer Synchronizer
logger logging.LoggerInterface
config conf.AdvancedConfig
pushManager push.Manager
managerStatus chan int
streamingStatus chan int64
operationMode int32
lifecycle lifecycle.Manager
backoff backoff.Interface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewSynchronizerManager creates new sync manager
@@ -57,22 +60,29 @@ func NewSynchronizerManager(
authClient service.AuthClient,
splitStorage storage.SplitStorage,
managerStatus chan int,
runtimeTelemetry storage.TelemetryRuntimeProducer,
metadata dtos.Metadata,
clientKey *string,
) (*ManagerImpl, error) {
if managerStatus == nil || cap(managerStatus) < 1 {
return nil, errors.New("Status channel cannot be nil nor having capacity")
}
manager := &ManagerImpl{
backoff: backoff.New(),
synchronizer: synchronizer,
logger: logger,
config: config,
managerStatus: managerStatus,
backoff: backoff.New(),
synchronizer: synchronizer,
logger: logger,
config: config,
managerStatus: managerStatus,
runtimeTelemetry: runtimeTelemetry,
}
manager.lifecycle.Setup()
if config.StreamingEnabled {
streamingStatus := make(chan int64, 1000)
pushManager, err := push.NewManager(logger, synchronizer, &config, streamingStatus, authClient)
if clientKey != nil && len(*clientKey) != 4 {
return nil, errors.New("invalid ClientKey")
}
pushManager, err := push.NewManager(logger, synchronizer, &config, streamingStatus, authClient, runtimeTelemetry, metadata, clientKey)
if err != nil {
return nil, err
}
@@ -182,6 +192,8 @@ func (s *ManagerImpl) pushStatusWatcher() {
s.pushManager.Stop()
s.synchronizer.SyncAll(false)
s.startPolling()
// Tracking STREAMING_DISABLED
s.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeStreamingStatus, telemetry.StreamingDisabled))
}
}
}
@@ -190,6 +202,8 @@ func (s *ManagerImpl) pushStatusWatcher() {
func (s *ManagerImpl) startPolling() {
atomic.StoreInt32(&s.operationMode, Polling)
s.synchronizer.StartPeriodicFetching()
// Tracking POLLING
s.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeSyncMode, telemetry.Polling))
}
func (s *ManagerImpl) stopPolling() {
@@ -198,10 +212,16 @@ func (s *ManagerImpl) stopPolling() {
func (s *ManagerImpl) pauseStreaming() {
s.pushManager.StartWorkers()
// Tracking STREAMING_PAUSED
s.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeStreamingStatus, telemetry.StreamingPaused))
}
func (s *ManagerImpl) enableStreaming() {
s.pushManager.StartWorkers()
atomic.StoreInt32(&s.operationMode, Streaming)
s.backoff.Reset()
// Tracking STREAMING
s.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeSyncMode, telemetry.Streaming))
// Tracking STREAMING_ENABLED
s.runtimeTelemetry.RecordStreamingEvent(telemetry.GetStreamingEvent(telemetry.EventTypeStreamingStatus, telemetry.StreamingEnabled))
}

9
vendor/github.com/splitio/go-split-commons/v3/synchronizer/synchronizer.go сгенерированный поставляемый
Просмотреть файл

@@ -5,10 +5,10 @@ import (
"github.com/splitio/go-split-commons/v3/synchronizer/worker/event"
"github.com/splitio/go-split-commons/v3/synchronizer/worker/impression"
"github.com/splitio/go-split-commons/v3/synchronizer/worker/impressionscount"
"github.com/splitio/go-split-commons/v3/synchronizer/worker/metric"
"github.com/splitio/go-split-commons/v3/synchronizer/worker/segment"
"github.com/splitio/go-split-commons/v3/synchronizer/worker/split"
"github.com/splitio/go-split-commons/v3/tasks"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/asynctask"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -27,7 +27,7 @@ type SplitTasks struct {
type Workers struct {
SplitFetcher split.Updater
SegmentFetcher segment.Updater
TelemetryRecorder metric.MetricRecorder
TelemetryRecorder telemetry.TelemetrySynchronizer
ImpressionRecorder impression.ImpressionRecorder
EventRecorder event.EventRecorder
ImpressionsCountRecorder impressionscount.ImpressionsCountRecorder
@@ -62,7 +62,7 @@ func NewSynchronizer(
}
func (s *SynchronizerImpl) dataFlusher() {
for true {
for {
msg := <-s.inMememoryFullQueue
switch msg {
case "EVENTS_FULL":
@@ -71,7 +71,6 @@ func (s *SynchronizerImpl) dataFlusher() {
if err != nil {
s.logger.Error("Error flushing storage queue", err)
}
break
case "IMPRESSIONS_FULL":
s.logger.Debug("FLUSHING storage queue")
err := s.workers.ImpressionRecorder.SynchronizeImpressions(s.impressionBulkSize)
@@ -137,7 +136,7 @@ func (s *SynchronizerImpl) StopPeriodicDataRecording() {
s.splitTasks.ImpressionSyncTask.Stop(true)
}
if s.splitTasks.TelemetrySyncTask != nil {
s.splitTasks.TelemetrySyncTask.Stop(false)
s.splitTasks.TelemetrySyncTask.Stop(true)
}
if s.splitTasks.EventSyncTask != nil {
s.splitTasks.EventSyncTask.Stop(true)

32
vendor/github.com/splitio/go-split-commons/v3/synchronizer/worker/event/single.go сгенерированный поставляемый
Просмотреть файл

@@ -2,39 +2,38 @@ package event
import (
"errors"
"strconv"
"time"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/util"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/logging"
)
// RecorderSingle struct for event sync
type RecorderSingle struct {
eventStorage storage.EventStorageConsumer
eventRecorder service.EventsRecorder
metricsWrapper *storage.MetricWrapper
logger logging.LoggerInterface
metadata dtos.Metadata
eventStorage storage.EventStorageConsumer
eventRecorder service.EventsRecorder
logger logging.LoggerInterface
metadata dtos.Metadata
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewEventRecorderSingle creates new event synchronizer for posting events
func NewEventRecorderSingle(
eventStorage storage.EventStorageConsumer,
eventRecorder service.EventsRecorder,
metricsWrapper *storage.MetricWrapper,
logger logging.LoggerInterface,
metadata dtos.Metadata,
runtimeTelemetry storage.TelemetryRuntimeProducer,
) EventRecorder {
return &RecorderSingle{
eventStorage: eventStorage,
eventRecorder: eventRecorder,
metricsWrapper: metricsWrapper,
logger: logger,
metadata: metadata,
eventStorage: eventStorage,
eventRecorder: eventRecorder,
logger: logger,
metadata: metadata,
runtimeTelemetry: runtimeTelemetry,
}
}
@@ -55,13 +54,12 @@ func (e *RecorderSingle) SynchronizeEvents(bulkSize int64) error {
err = e.eventRecorder.Record(queuedEvents, e.metadata)
if err != nil {
if httpError, ok := err.(*dtos.HTTPError); ok {
e.metricsWrapper.StoreCounters(storage.PostEventsCounter, strconv.Itoa(httpError.Code))
e.runtimeTelemetry.RecordSyncError(telemetry.EventSync, httpError.Code)
}
return err
}
bucket := util.Bucket(time.Now().Sub(before).Nanoseconds())
e.metricsWrapper.StoreLatencies(storage.PostEventsLatency, bucket)
e.metricsWrapper.StoreCounters(storage.PostEventsCounter, "ok")
e.runtimeTelemetry.RecordSyncLatency(telemetry.EventSync, time.Since(before).Nanoseconds())
e.runtimeTelemetry.RecordSuccessfulSync(telemetry.EventSync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return nil
}

15
vendor/github.com/splitio/go-split-commons/v3/synchronizer/worker/impression/single.go сгенерированный поставляемый
Просмотреть файл

@@ -2,13 +2,13 @@ package impression
import (
"errors"
"strconv"
"time"
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-split-commons/v3/util"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -22,20 +22,20 @@ const (
type RecorderSingle struct {
impressionStorage storage.ImpressionStorageConsumer
impressionRecorder service.ImpressionsRecorder
metricsWrapper *storage.MetricWrapper
logger logging.LoggerInterface
metadata dtos.Metadata
mode string
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewRecorderSingle creates new impression synchronizer for posting impressions
func NewRecorderSingle(
impressionStorage storage.ImpressionStorageConsumer,
impressionRecorder service.ImpressionsRecorder,
metricsWrapper *storage.MetricWrapper,
logger logging.LoggerInterface,
metadata dtos.Metadata,
managerConfig conf.ManagerConfig,
runtimeTelemetry storage.TelemetryRuntimeProducer,
) ImpressionRecorder {
mode := conf.ImpressionsModeOptimized
if !util.ShouldBeOptimized(managerConfig) {
@@ -44,10 +44,10 @@ func NewRecorderSingle(
return &RecorderSingle{
impressionStorage: impressionStorage,
impressionRecorder: impressionRecorder,
metricsWrapper: metricsWrapper,
logger: logger,
metadata: metadata,
mode: mode,
runtimeTelemetry: runtimeTelemetry,
}
}
@@ -96,13 +96,12 @@ func (i *RecorderSingle) SynchronizeImpressions(bulkSize int64) error {
err = i.impressionRecorder.Record(bulkImpressions, i.metadata, map[string]string{splitSDKImpressionsMode: i.mode})
if err != nil {
if httpError, ok := err.(*dtos.HTTPError); ok {
i.metricsWrapper.StoreCounters(storage.TestImpressionsCounter, strconv.Itoa(httpError.Code))
i.runtimeTelemetry.RecordSyncError(telemetry.ImpressionSync, httpError.Code)
}
return err
}
bucket := util.Bucket(time.Now().Sub(before).Nanoseconds())
i.metricsWrapper.StoreLatencies(storage.TestImpressionsLatency, bucket)
i.metricsWrapper.StoreCounters(storage.TestImpressionsCounter, "ok")
i.runtimeTelemetry.RecordSyncLatency(telemetry.ImpressionSync, time.Since(before).Nanoseconds())
i.runtimeTelemetry.RecordSuccessfulSync(telemetry.ImpressionSync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return nil
}

Просмотреть файл

@@ -1,9 +1,13 @@
package impressionscount
import (
"time"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/provisional"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -13,6 +17,7 @@ type RecorderSingle struct {
impressionRecorder service.ImpressionsRecorder
metadata dtos.Metadata
logger logging.LoggerInterface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewRecorderSingle creates new impressionsCount synchronizer for posting impressionsCount
@@ -21,12 +26,14 @@ func NewRecorderSingle(
impressionRecorder service.ImpressionsRecorder,
metadata dtos.Metadata,
logger logging.LoggerInterface,
runtimeTelemetry storage.TelemetryRuntimeProducer,
) ImpressionsCountRecorder {
return &RecorderSingle{
impressionsCounter: impressionsCounter,
impressionRecorder: impressionRecorder,
metadata: metadata,
logger: logger,
runtimeTelemetry: runtimeTelemetry,
}
}
@@ -47,5 +54,16 @@ func (m *RecorderSingle) SynchronizeImpressionsCount() error {
pf := dtos.ImpressionsCountDTO{
PerFeature: impressionsInTimeFrame,
}
return m.impressionRecorder.RecordImpressionsCount(pf, m.metadata)
before := time.Now()
err := m.impressionRecorder.RecordImpressionsCount(pf, m.metadata)
if err != nil {
if httpError, ok := err.(*dtos.HTTPError); ok {
m.runtimeTelemetry.RecordSyncError(telemetry.ImpressionCountSync, httpError.Code)
}
return err
}
m.runtimeTelemetry.RecordSyncLatency(telemetry.ImpressionCountSync, time.Since(before).Nanoseconds())
m.runtimeTelemetry.RecordSuccessfulSync(telemetry.ImpressionCountSync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return nil
}

6
vendor/github.com/splitio/go-split-commons/v3/synchronizer/worker/metric/interface.go сгенерированный поставляемый
Просмотреть файл

@@ -1,6 +0,0 @@
package metric
// MetricRecorder interface
type MetricRecorder interface {
SynchronizeTelemetry() error
}

74
vendor/github.com/splitio/go-split-commons/v3/synchronizer/worker/metric/single.go сгенерированный поставляемый
Просмотреть файл

@@ -1,74 +0,0 @@
package metric
import (
"errors"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
)
// RecorderSingle struct for metric sync
type RecorderSingle struct {
metricStorage storage.MetricsStorageConsumer
metricRecorder service.MetricsRecorder
metadata dtos.Metadata
}
// NewRecorderSingle creates new metric synchronizer for posting metrics
func NewRecorderSingle(
metricStorage storage.MetricsStorageConsumer,
metricRecorder service.MetricsRecorder,
metadata dtos.Metadata,
) MetricRecorder {
return &RecorderSingle{
metricStorage: metricStorage,
metricRecorder: metricRecorder,
metadata: metadata,
}
}
func (m *RecorderSingle) synchronizeLatencies() error {
latencies := m.metricStorage.PopLatencies()
if len(latencies) > 0 {
err := m.metricRecorder.RecordLatencies(latencies, m.metadata)
return err
}
return nil
}
func (m *RecorderSingle) synchronizeGauges() error {
var errs []error
for _, gauge := range m.metricStorage.PopGauges() {
err := m.metricRecorder.RecordGauge(gauge, m.metadata)
if err != nil {
errs = append(errs, err)
}
}
if len(errs) > 0 {
return errors.New("Some gauges could not be posted")
}
return nil
}
func (m *RecorderSingle) synchronizeCounters() error {
counters := m.metricStorage.PopCounters()
if len(counters) > 0 {
err := m.metricRecorder.RecordCounters(counters, m.metadata)
return err
}
return nil
}
// SynchronizeTelemetry syncs telemetry
func (m *RecorderSingle) SynchronizeTelemetry() error {
err := m.synchronizeGauges()
if err != nil {
return err
}
err = m.synchronizeLatencies()
if err != nil {
return err
}
return m.synchronizeCounters()
}

33
vendor/github.com/splitio/go-split-commons/v3/synchronizer/worker/segment/segment.go сгенерированный поставляемый
Просмотреть файл

@@ -2,25 +2,24 @@ package segment
import (
"fmt"
"strconv"
"sync"
"time"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/util"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/datastructures/set"
"github.com/splitio/go-toolkit/v4/logging"
)
// UpdaterImpl struct for segment sync
type UpdaterImpl struct {
splitStorage storage.SplitStorageConsumer
segmentStorage storage.SegmentStorage
segmentFetcher service.SegmentFetcher
metricsWrapper *storage.MetricWrapper
logger logging.LoggerInterface
splitStorage storage.SplitStorageConsumer
segmentStorage storage.SegmentStorage
segmentFetcher service.SegmentFetcher
logger logging.LoggerInterface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewSegmentFetcher creates new segment synchronizer for processing segment updates
@@ -28,15 +27,15 @@ func NewSegmentFetcher(
splitStorage storage.SplitStorage,
segmentStorage storage.SegmentStorage,
segmentFetcher service.SegmentFetcher,
metricsWrapper *storage.MetricWrapper,
logger logging.LoggerInterface,
runtimeTelemetry storage.TelemetryRuntimeProducer,
) Updater {
return &UpdaterImpl{
splitStorage: splitStorage,
segmentStorage: segmentStorage,
segmentFetcher: segmentFetcher,
metricsWrapper: metricsWrapper,
logger: logger,
splitStorage: splitStorage,
segmentStorage: segmentStorage,
segmentFetcher: segmentFetcher,
logger: logger,
runtimeTelemetry: runtimeTelemetry,
}
}
@@ -83,16 +82,14 @@ func (s *UpdaterImpl) SynchronizeSegment(name string, till *int64, requestNoCach
segmentChanges, err := s.segmentFetcher.Fetch(name, changeNumber, requestNoCache)
if err != nil {
if httpError, ok := err.(*dtos.HTTPError); ok {
s.metricsWrapper.StoreCounters(storage.SegmentChangesCounter, strconv.Itoa(httpError.Code))
s.runtimeTelemetry.RecordSyncError(telemetry.SegmentSync, httpError.Code)
}
return err
}
s.runtimeTelemetry.RecordSyncLatency(telemetry.SegmentSync, time.Since(before).Nanoseconds())
s.processUpdate(segmentChanges)
bucket := util.Bucket(time.Now().Sub(before).Nanoseconds())
s.metricsWrapper.StoreLatencies(storage.SegmentChangesLatency, bucket)
s.metricsWrapper.StoreCounters(storage.SegmentChangesCounter, "ok")
if segmentChanges.Till == segmentChanges.Since || (till != nil && segmentChanges.Till >= *till) {
s.runtimeTelemetry.RecordSuccessfulSync(telemetry.SegmentSync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return nil
}
}

30
vendor/github.com/splitio/go-split-commons/v3/synchronizer/worker/split/split.go сгенерированный поставляемый
Просмотреть файл

@@ -1,13 +1,12 @@
package split
import (
"strconv"
"time"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-split-commons/v3/util"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/logging"
)
@@ -17,24 +16,24 @@ const (
// UpdaterImpl struct for split sync
type UpdaterImpl struct {
splitStorage storage.SplitStorage
splitFetcher service.SplitFetcher
metricsWrapper *storage.MetricWrapper
logger logging.LoggerInterface
splitStorage storage.SplitStorage
splitFetcher service.SplitFetcher
logger logging.LoggerInterface
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewSplitFetcher creates new split synchronizer for processing split updates
func NewSplitFetcher(
splitStorage storage.SplitStorage,
splitFetcher service.SplitFetcher,
metricsWrapper *storage.MetricWrapper,
logger logging.LoggerInterface,
) *UpdaterImpl {
runtimeTelemetry storage.TelemetryRuntimeProducer,
) Updater {
return &UpdaterImpl{
splitStorage: splitStorage,
splitFetcher: splitFetcher,
metricsWrapper: metricsWrapper,
logger: logger,
splitStorage: splitStorage,
splitFetcher: splitFetcher,
logger: logger,
runtimeTelemetry: runtimeTelemetry,
}
}
@@ -76,16 +75,15 @@ func (s *UpdaterImpl) SynchronizeSplits(till *int64, requestNoCache bool) ([]str
splits, err := s.splitFetcher.Fetch(changeNumber, requestNoCache)
if err != nil {
if httpError, ok := err.(*dtos.HTTPError); ok {
s.metricsWrapper.StoreCounters(storage.SplitChangesCounter, strconv.Itoa(httpError.Code))
s.runtimeTelemetry.RecordSyncError(telemetry.SplitSync, httpError.Code)
}
return segments, err
}
s.runtimeTelemetry.RecordSyncLatency(telemetry.SplitSync, time.Since(before).Nanoseconds())
s.processUpdate(splits)
segments = append(segments, extractSegments(splits)...)
bucket := util.Bucket(time.Now().Sub(before).Nanoseconds())
s.metricsWrapper.StoreCounters(storage.SplitChangesCounter, "ok")
s.metricsWrapper.StoreLatencies(storage.SplitChangesLatency, bucket)
if splits.Till == splits.Since || (till != nil && splits.Till >= *till) {
s.runtimeTelemetry.RecordSuccessfulSync(telemetry.SplitSync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return segments, nil
}
}

Просмотреть файл

@@ -1,19 +1,19 @@
package tasks
import (
"github.com/splitio/go-split-commons/v3/synchronizer/worker/metric"
"github.com/splitio/go-split-commons/v3/telemetry"
"github.com/splitio/go-toolkit/v4/asynctask"
"github.com/splitio/go-toolkit/v4/logging"
)
// NewRecordTelemetryTask creates a new telemtry recording task
func NewRecordTelemetryTask(
recorder metric.MetricRecorder,
recorder telemetry.TelemetrySynchronizer,
period int,
logger logging.LoggerInterface,
) *asynctask.AsyncTask {
record := func(logger logging.LoggerInterface) error {
return recorder.SynchronizeTelemetry()
return recorder.SynchronizeStats()
}
onStop := func(l logging.LoggerInterface) {

107
vendor/github.com/splitio/go-split-commons/v3/telemetry/constants.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,107 @@
package telemetry
import "github.com/splitio/go-split-commons/v3/conf"
const (
// Treatment getTreatment
Treatment = "treatment"
// Treatments getTreatments
Treatments = "treatments"
// TreatmentWithConfig getTreatmentWithConfig
TreatmentWithConfig = "treatmentWithConfig"
// TreatmentsWithConfig getTreatmentsWithConfig
TreatmentsWithConfig = "treatmentsWithConfig"
// Track track
Track = "track"
)
const (
// SplitSync splitChanges
SplitSync = iota
// SegmentSync segmentChanges
SegmentSync
// ImpressionSync impressions
ImpressionSync
// ImpressionCountSync impressionsCount
ImpressionCountSync
// EventSync events
EventSync
// TelemetrySync telemetry
TelemetrySync
// TokenSync auth
TokenSync
)
const (
// ImpressionsDropped dropped
ImpressionsDropped = iota
// ImpressionsDeduped deduped
ImpressionsDeduped
// ImpressionsQueued queued
ImpressionsQueued
)
const (
// EventsDropped dropped
EventsDropped = iota
// EventsQueued queued
EventsQueued
)
const (
// LatencyBucketCount Max buckets
LatencyBucketCount = 23
// MaxStreamingEvents Max streaming events allowed
MaxStreamingEvents = 20
// MaxTags Max tags
MaxTags = 10
)
const (
EventTypeSSEConnectionEstablished = iota * 10
EventTypeOccupancyPri
EventTypeOccupancySec
EventTypeStreamingStatus
EventTypeConnectionError
EventTypeTokenRefresh
EventTypeAblyError
EventTypeSyncMode
)
const (
StreamingDisabled = iota
StreamingEnabled
StreamingPaused
)
const (
Requested = iota
NonRequested
)
const (
Streaming = iota
Polling
)
const (
Standalone = iota
Consumer
)
const (
ImpressionsModeOptimized = iota
ImpressionsModeDebug
)
const (
Redis = "redis"
Memory = "memory"
)
// InitConfig involves entire config for init
type InitConfig struct {
AdvancedConfig conf.AdvancedConfig
TaskPeriods conf.TaskPeriods
ManagerConfig conf.ManagerConfig
}

42
vendor/github.com/splitio/go-split-commons/v3/telemetry/helpers.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,42 @@
package telemetry
import (
"time"
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
)
func GetStreamingEvent(eventType int, data int64) *dtos.StreamingEvent {
switch eventType {
case EventTypeSSEConnectionEstablished, EventTypeOccupancyPri,
EventTypeOccupancySec, EventTypeStreamingStatus,
EventTypeConnectionError, EventTypeTokenRefresh,
EventTypeAblyError, EventTypeSyncMode:
return &dtos.StreamingEvent{
Type: eventType,
Data: data,
Timestamp: time.Now().UTC().Unix(),
}
}
return nil
}
func getURLOverrides(cfg conf.AdvancedConfig) dtos.URLOverrides {
defaults := conf.GetDefaultAdvancedConfig()
return dtos.URLOverrides{
Sdk: cfg.SdkURL != defaults.SdkURL,
Events: cfg.EventsURL != defaults.EventsURL,
Auth: cfg.AuthServiceURL != defaults.AuthServiceURL,
Stream: cfg.StreamingServiceURL != defaults.StreamingServiceURL,
Telemetry: cfg.TelemetryServiceURL != defaults.TelemetryServiceURL,
}
}
func getRedudantActiveFactories(factoryInstances map[string]int64) int64 {
var toReturn int64 = 0
for _, instances := range factoryInstances {
toReturn = toReturn + instances - 1
}
return toReturn
}

7
vendor/github.com/splitio/go-split-commons/v3/telemetry/interface.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,7 @@
package telemetry
// TelemetrySynchronizer interface
type TelemetrySynchronizer interface {
SynchronizeConfig(cfg InitConfig, timedUntilReady int64, factoryInstances map[string]int64, tags []string)
SynchronizeStats() error
}

135
vendor/github.com/splitio/go-split-commons/v3/telemetry/memory.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,135 @@
package telemetry
import (
"os"
"strings"
"time"
"github.com/splitio/go-split-commons/v3/conf"
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/service"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-toolkit/v4/logging"
)
// RecorderSingle struct for telemetry sync
type RecorderSingle struct {
telemetryStorage storage.TelemetryStorageConsumer
telemetryRecorder service.TelemetryRecorder
splitStorage storage.SplitStorageConsumer
segmentStorage storage.SegmentStorageConsumer
logger logging.LoggerInterface
metadata dtos.Metadata
runtimeTelemetry storage.TelemetryRuntimeProducer
}
// NewTelemetrySynchronizer creates new event synchronizer for posting events
func NewTelemetrySynchronizer(
telemetryStorage storage.TelemetryStorageConsumer,
telemetryRecorder service.TelemetryRecorder,
splitStorage storage.SplitStorageConsumer,
segmentStorage storage.SegmentStorageConsumer,
logger logging.LoggerInterface,
metadata dtos.Metadata,
runtimeTelemetry storage.TelemetryRuntimeProducer,
) TelemetrySynchronizer {
return &RecorderSingle{
telemetryStorage: telemetryStorage,
telemetryRecorder: telemetryRecorder,
splitStorage: splitStorage,
segmentStorage: segmentStorage,
logger: logger,
metadata: metadata,
runtimeTelemetry: runtimeTelemetry,
}
}
func (e *RecorderSingle) buildStats() dtos.Stats {
methodLatencies := e.telemetryStorage.PopLatencies()
methodExceptions := e.telemetryStorage.PopExceptions()
lastSynchronization := e.telemetryStorage.GetLastSynchronization()
httpErrors := e.telemetryStorage.PopHTTPErrors()
httpLatencies := e.telemetryStorage.PopHTTPLatencies()
return dtos.Stats{
MethodLatencies: &methodLatencies,
MethodExceptions: &methodExceptions,
ImpressionsDropped: e.telemetryStorage.GetImpressionsStats(ImpressionsDropped),
ImpressionsDeduped: e.telemetryStorage.GetImpressionsStats(ImpressionsDeduped),
ImpressionsQueued: e.telemetryStorage.GetImpressionsStats(ImpressionsQueued),
EventsQueued: e.telemetryStorage.GetEventsStats(EventsQueued),
EventsDropped: e.telemetryStorage.GetEventsStats(EventsDropped),
LastSynchronizations: &lastSynchronization,
HTTPErrors: &httpErrors,
HTTPLatencies: &httpLatencies,
SplitCount: int64(len(e.splitStorage.SplitNames())),
SegmentCount: int64(e.splitStorage.SegmentNames().Size()),
SegmentKeyCount: e.segmentStorage.SegmentKeysCount(),
TokenRefreshes: e.telemetryStorage.PopTokenRefreshes(),
AuthRejections: e.telemetryStorage.PopAuthRejections(),
StreamingEvents: e.telemetryStorage.PopStreamingEvents(),
SessionLengthMs: e.telemetryStorage.GetSessionLength(),
Tags: e.telemetryStorage.PopTags(),
}
}
// SynchronizeStats syncs telemetry stats
func (e *RecorderSingle) SynchronizeStats() error {
stats := e.buildStats()
before := time.Now()
err := e.telemetryRecorder.RecordStats(stats, e.metadata)
if err != nil {
if httpError, ok := err.(*dtos.HTTPError); ok {
e.runtimeTelemetry.RecordSyncError(TelemetrySync, httpError.Code)
}
return err
}
e.runtimeTelemetry.RecordSyncLatency(TelemetrySync, time.Since(before).Nanoseconds())
e.runtimeTelemetry.RecordSuccessfulSync(TelemetrySync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
return nil
}
// SynchronizeConfig syncs telemetry config
func (e *RecorderSingle) SynchronizeConfig(cfg InitConfig, timedUntilReady int64, factoryInstances map[string]int64, tags []string) {
urlOverrides := getURLOverrides(cfg.AdvancedConfig)
impressionsMode := ImpressionsModeOptimized
if cfg.ManagerConfig.ImpressionsMode == conf.ImpressionsModeDebug {
impressionsMode = ImpressionsModeDebug
}
before := time.Now()
err := e.telemetryRecorder.RecordConfig(dtos.Config{
OperationMode: Standalone,
Storage: Memory,
ActiveFactories: int64(len(factoryInstances)),
RedundantFactories: getRedudantActiveFactories(factoryInstances),
Tags: tags,
StreamingEnabled: cfg.AdvancedConfig.StreamingEnabled,
Rates: &dtos.Rates{
Splits: int64(cfg.TaskPeriods.SplitSync),
Segments: int64(cfg.TaskPeriods.SegmentSync),
Impressions: int64(cfg.TaskPeriods.ImpressionSync),
Events: int64(cfg.TaskPeriods.EventsSync),
Telemetry: int64(cfg.TaskPeriods.TelemetrySync),
},
URLOverrides: &urlOverrides,
ImpressionsQueueSize: int64(cfg.AdvancedConfig.ImpressionsQueueSize),
EventsQueueSize: int64(cfg.AdvancedConfig.EventsQueueSize),
ImpressionsMode: impressionsMode,
ImpressionsListenerEnabled: cfg.ManagerConfig.ListenerEnabled,
HTTPProxyDetected: len(strings.TrimSpace(os.Getenv("HTTP_PROXY"))) > 0,
TimeUntilReady: timedUntilReady,
BurTimeouts: e.telemetryStorage.GetBURTimeouts(),
NonReadyUsages: e.telemetryStorage.GetNonReadyUsages(),
}, e.metadata)
if err != nil {
e.logger.Error("Could not log config data", err.Error())
if httpError, ok := err.(*dtos.HTTPError); ok {
e.runtimeTelemetry.RecordSyncError(TelemetrySync, httpError.Code)
}
return
}
e.runtimeTelemetry.RecordSyncLatency(TelemetrySync, time.Since(before).Nanoseconds())
e.runtimeTelemetry.RecordSuccessfulSync(TelemetrySync, time.Now().UTC().UnixNano()/int64(time.Millisecond))
}

Просмотреть файл

@@ -1,4 +1,4 @@
package util
package telemetry
var latencyBuckets = [23]float64{
1.00,

37
vendor/github.com/splitio/go-split-commons/v3/telemetry/redis.go сгенерированный поставляемый Обычный файл
Просмотреть файл

@@ -0,0 +1,37 @@
package telemetry
import (
"github.com/splitio/go-split-commons/v3/dtos"
"github.com/splitio/go-split-commons/v3/storage"
"github.com/splitio/go-toolkit/v4/logging"
)
type SynchronizerRedis struct {
storage storage.TelemetryConfigProducer
logger logging.LoggerInterface
}
func NewSynchronizerRedis(storage storage.TelemetryConfigProducer, logger logging.LoggerInterface) TelemetrySynchronizer {
return &SynchronizerRedis{
storage: storage,
logger: logger,
}
}
func (r *SynchronizerRedis) SynchronizeStats() error {
// No-Op. Not required for redis. This will be implemented by Synchronizer.
return nil
}
func (r *SynchronizerRedis) SynchronizeConfig(cfg InitConfig, timedUntilReady int64, factoryInstances map[string]int64, tags []string) {
err := r.storage.RecordConfigData(dtos.Config{
OperationMode: Consumer,
Storage: Redis,
ActiveFactories: int64(len(factoryInstances)),
RedundantFactories: getRedudantActiveFactories(factoryInstances),
Tags: tags,
})
if err != nil {
r.logger.Error("Could not log config data", err.Error())
}
}