Merge pull request #126 from deploymenttheory/dev

Refactor

Author: ShocOne

concurrenyhandler/dynamic_token_adjustment.go

@@ -0,0 +1,199 @@
+// concurrencyhandler/dynamic_token_adjustment.go
+
+package concurrencyhandler
+
+import (
+	"time"
+
+	"go.uber.org/zap"
+)
+
+// AdjustConcurrencyLimit dynamically modifies the maximum concurrency limit
+// based on the newLimit provided. This function helps in adjusting the concurrency
+// limit in real-time based on observed system performance and other metrics. It
+// transfers the tokens from the old semaphore to the new one, ensuring that there's
+// no loss of tokens during the transition.
+func (ch *ConcurrencyHandler) AdjustConcurrencyLimit(newLimit int) {
+	ch.lock.Lock()
+	defer ch.lock.Unlock()
+
+	if newLimit <= 0 {
+		return // Avoid setting a non-positive limit
+	}
+
+	// Create a new semaphore with the desired limit
+	newSem := make(chan struct{}, newLimit)
+
+	// Transfer tokens from the old semaphore to the new one
+	for i := 0; i < len(ch.sem) && i < newLimit; i++ {
+		newSem <- struct{}{}
+	}
+
+	ch.sem = newSem
+}
+
+// AdjustConcurrencyBasedOnMetrics evaluates the current metrics and adjusts the
+// concurrency limit if required. It checks metrics like average token acquisition
+// time and decides on a new concurrency limit. The method ensures that the new
+// limit respects the minimum and maximum allowed concurrency bounds.
+func (ch *ConcurrencyHandler) AdjustConcurrencyBasedOnMetrics() {
+	// Calculate the average acquisition time
+	avgAcquisitionTime := ch.AverageAcquisitionTime()
+
+	// Get the current concurrency limit
+	currentLimit := cap(ch.sem)
+
+	// Calculate the historical average acquisition time
+	historicalAvgAcquisitionTime := ch.HistoricalAverageAcquisitionTime()
+
+	// Decide on a new limit based on metrics
+	newLimit := currentLimit
+	if avgAcquisitionTime > time.Duration(float64(historicalAvgAcquisitionTime)*1.2) { // 20% increase in acquisition time
+		newLimit = currentLimit - 2 // decrease concurrency more aggressively
+	} else if avgAcquisitionTime < time.Duration(float64(historicalAvgAcquisitionTime)*0.8) { // 20% decrease in acquisition time
+		newLimit = currentLimit + 2 // increase concurrency more aggressively
+	} else if avgAcquisitionTime > historicalAvgAcquisitionTime {
+		newLimit = currentLimit - 1 // decrease concurrency conservatively
+	} else if avgAcquisitionTime < historicalAvgAcquisitionTime {
+		newLimit = currentLimit + 1 // increase concurrency conservatively
+	}
+
+	// Ensure newLimit is within safety bounds
+	if newLimit > MaxConcurrency {
+		newLimit = MaxConcurrency
+	} else if newLimit < MinConcurrency {
+		newLimit = MinConcurrency
+	}
+
+	// Adjust concurrency if the new limit is different from the current
+	if newLimit != currentLimit {
+		ch.AdjustConcurrencyLimit(newLimit)
+
+		// Log the adjustment
+		ch.logger.Debug("Adjusted concurrency",
+			zap.Int("OldLimit", currentLimit),
+			zap.Int("NewLimit", newLimit),
+			zap.String("Reason", "Based on average acquisition time"),
+			zap.Duration("AverageAcquisitionTime", avgAcquisitionTime),
+			zap.Duration("HistoricalAverageAcquisitionTime", historicalAvgAcquisitionTime),
+		)
+	}
+}
+
+// EvaluateMetricsAndAdjustConcurrency evaluates the performance metrics and makes necessary
+// adjustments to the concurrency limit. The method assesses the average response time
+// and adjusts the concurrency based on how it compares to the historical average acquisition time.
+// If the average response time has significantly increased compared to the historical average,
+// the concurrency limit is decreased, and vice versa. The method ensures that the concurrency
+// limit remains within the bounds defined by the system's best practices.
+func (ch *ConcurrencyHandler) EvaluateMetricsAndAdjustConcurrency() {
+	ch.PerfMetrics.lock.Lock()
+	averageResponseTime := ch.PerfMetrics.TotalResponseTime / time.Duration(ch.PerfMetrics.TotalRequests)
+	ch.PerfMetrics.lock.Unlock()
+
+	historicalAverageAcquisitionTime := ch.HistoricalAverageAcquisitionTime()
+
+	// Decide on the new limit based on the average response time compared to the historical average
+	newLimit := cap(ch.sem) // Start with the current limit
+	if averageResponseTime > time.Duration(float64(historicalAverageAcquisitionTime)*1.2) {
+		// Decrease concurrency more aggressively if the average response time has significantly increased
+		newLimit -= 1
+	} else if averageResponseTime < time.Duration(float64(historicalAverageAcquisitionTime)*0.8) {
+		// Increase concurrency more aggressively if the average response time has significantly decreased
+		newLimit += 1
+	}
+
+	// Ensure the new limit is within the defined bounds
+	if newLimit > MaxConcurrency {
+		newLimit = MaxConcurrency
+	} else if newLimit < MinConcurrency {
+		newLimit = MinConcurrency
+	}
+
+	// Adjust the concurrency limit if the new limit is different from the current limit
+	currentLimit := cap(ch.sem)
+	if newLimit != currentLimit {
+		ch.AdjustConcurrencyLimit(newLimit)
+
+		// Log the adjustment for debugging purposes
+		ch.logger.Debug("Adjusted concurrency",
+			zap.Int("OldLimit", currentLimit),
+			zap.Int("NewLimit", newLimit),
+			zap.String("Reason", "Based on evaluation of metrics"),
+			zap.Duration("AverageResponseTime", averageResponseTime),
+			zap.Duration("HistoricalAverageAcquisitionTime", historicalAverageAcquisitionTime),
+		)
+	}
+}
+
+//------ Concurrency Monitoring Functions:
+
+// StartMetricEvaluation continuously monitors the client's interactions with the API and adjusts the concurrency limits dynamically.
+// The function evaluates metrics at regular intervals to detect burst activity patterns.
+// If a burst activity is detected (e.g., many requests in a short period), the evaluation interval is reduced for more frequent checks.
+// Otherwise, it reverts to a default interval for regular checks.
+// After each evaluation, the function calls EvaluateMetricsAndAdjustConcurrency to potentially adjust the concurrency based on observed metrics.
+//
+// The evaluation process works as follows:
+// 1. Sleep for the defined evaluation interval.
+// 2. Check if there's a burst in activity using the isBurstActivity method.
+// 3. If a burst is detected, the evaluation interval is shortened to more frequently monitor and adjust the concurrency.
+// 4. If no burst is detected, it maintains the default evaluation interval.
+// 5. It then evaluates the metrics and adjusts the concurrency accordingly.
+func (ch *ConcurrencyHandler) StartMetricEvaluation() {
+	evalInterval := 5 * time.Minute // Initial interval
+
+	for {
+		time.Sleep(evalInterval) // Wait for the defined evaluation interval
+
+		// Determine if there's been a burst in activity
+		if ch.isBurstActivity() {
+			evalInterval = 1 * time.Minute // Increase the frequency of checks during burst activity
+		} else {
+			evalInterval = 5 * time.Minute // Revert to the default interval outside of burst periods
+		}
+
+		// Evaluate the current metrics and adjust concurrency if necessary
+		ch.EvaluateMetricsAndAdjustConcurrency()
+	}
+}
+
+// isBurstActivity checks if there's been a burst in activity based on the acquisition times of the tokens.
+// A burst is considered to have occurred if the time since the last token acquisition is short.
+func (ch *ConcurrencyHandler) isBurstActivity() bool {
+	// Lock before checking the last token acquisition time
+	ch.lock.Lock()
+	defer ch.lock.Unlock()
+
+	// Consider it a burst if the last token was acquired less than 2 minutes ago
+	return time.Since(ch.lastTokenAcquisitionTime) < 2*time.Minute
+}
+
+// StartConcurrencyAdjustment launches a periodic checker that evaluates current metrics and adjusts concurrency limits if needed.
+// It uses a ticker to periodically trigger the adjustment logich.
+func (ch *ConcurrencyHandler) StartConcurrencyAdjustment() {
+	ticker := time.NewTicker(EvaluationInterval)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		ch.AdjustConcurrencyBasedOnMetrics()
+	}
+}
+
+// Returns the average Acquisition Time to get a token from the semaphore
+func (ch *ConcurrencyHandler) GetAverageAcquisitionTime() time.Duration {
+	// Assuming ConcurrencyMgr has a method to get this metric
+	return ch.AverageAcquisitionTime()
+}
+
+func (ch *ConcurrencyHandler) GetHistoricalAverageAcquisitionTime() time.Duration {
+	// Assuming ConcurrencyMgr has a method to get this metric
+	return ch.HistoricalAverageAcquisitionTime()
+}
+
+// GetPerformanceMetrics returns the current performance metrics of the ConcurrencyHandler.
+// This includes counts of total requests, retries, rate limit errors, total response time,
+// and token wait time.
+func (ch *ConcurrencyHandler) GetPerformanceMetrics() *PerformanceMetrics {
+	return ch.PerfMetrics
+}

concurrenyhandler/handler.go

@@ -0,0 +1,59 @@
+// concurrencyhandler/handler.go
+package concurrencyhandler
+
+import (
+	"sync"
+	"time"
+
+	"github.com/deploymenttheory/go-api-http-client/logger"
+)
+
+// Constants and Data Structures:
+const (
+	MaxConcurrency     = 10              // Maximum allowed concurrent requests
+	MinConcurrency     = 1               // Minimum allowed concurrent requests
+	EvaluationInterval = 1 * time.Minute // Time interval for evaluating metrics and adjusting concurrency
+)
+
+// ConcurrencyHandler controls the number of concurrent HTTP requests.
+type ConcurrencyHandler struct {
+	sem                      chan struct{}
+	logger                   logger.Logger
+	AcquisitionTimes         []time.Duration
+	lock                     sync.Mutex
+	lastTokenAcquisitionTime time.Time
+	PerfMetrics              *PerformanceMetrics
+}
+
+// PerformanceMetrics captures various metrics related to the client's
+// interactions with the API.
+type PerformanceMetrics struct {
+	TotalRequests        int64
+	TotalRetries         int64
+	TotalRateLimitErrors int64
+	TotalResponseTime    time.Duration
+	TokenWaitTime        time.Duration
+	lock                 sync.Mutex // Protects performance metrics fields
+}
+
+// NewConcurrencyManager initializes a new ConcurrencyManager with the given
+// concurrency limit, logger, and perf metrics. The ConcurrencyManager ensures
+// no more than a certain number of concurrent requests are made.
+// It uses a semaphore to control concurrency.
+func NewConcurrencyHandler(limit int, logger logger.Logger, perfMetrics *PerformanceMetrics) *ConcurrencyHandler {
+	return &ConcurrencyHandler{
+		sem:              make(chan struct{}, limit),
+		logger:           logger,
+		AcquisitionTimes: []time.Duration{},
+		PerfMetrics:      perfMetrics,
+	}
+}
+
+// requestIDKey is type used as a key for storing and retrieving
+// request-specific identifiers from a context.Context object. This private
+// type ensures that the key is distinct and prevents accidental value
+// retrieval or conflicts with other context keys. The value associated
+// with this key in a context is typically a UUID that uniquely identifies
+// a request being processed by the ConcurrencyManager, allowing for
+// fine-grained control and tracking of concurrent HTTP requests.
+type requestIDKey struct{}

concurrenyhandler/metrics.go

@@ -0,0 +1,69 @@
+// concurrencyhandler/metrics.go
+package concurrencyhandler
+
+import "time"
+
+// AverageAcquisitionTime computes the average time taken to acquire a token
+// from the semaphore. It helps in understanding the contention for tokens
+// and can be used to adjust concurrency limits.
+func (ch *ConcurrencyHandler) AverageAcquisitionTime() time.Duration {
+	ch.lock.Lock()
+	defer ch.lock.Unlock()
+
+	if len(ch.AcquisitionTimes) == 0 {
+		return 0
+	}
+
+	totalTime := time.Duration(0)
+	for _, t := range ch.AcquisitionTimes {
+		totalTime += t
+	}
+	return totalTime / time.Duration(len(ch.AcquisitionTimes))
+}
+
+// HistoricalAverageAcquisitionTime computes the average time taken to acquire
+// a token from the semaphore over a historical period (e.g., the last 5 minutes).
+// It helps in understanding the historical contention for tokens and can be used
+// to adjust concurrency limits.
+func (ch *ConcurrencyHandler) HistoricalAverageAcquisitionTime() time.Duration {
+	ch.lock.Lock()
+	defer ch.lock.Unlock()
+
+	// For simplicity, let's say we store the last 5 minutes of acquisition times.
+	// This means if EvaluationInterval is 1 minute, we consider the last 5 data points.
+	historicalCount := 5
+	if len(ch.AcquisitionTimes) < historicalCount {
+		return ch.AverageAcquisitionTime() // If not enough historical data, return the overall average
+	}
+
+	totalTime := time.Duration(0)
+	for _, t := range ch.AcquisitionTimes[len(ch.AcquisitionTimes)-historicalCount:] {
+		totalTime += t
+	}
+	return totalTime / time.Duration(historicalCount)
+}
+
+// updatePerformanceMetrics updates the ConcurrencyHandler's performance metrics
+// by recording the duration of an HTTP request and incrementing the total
+// request count. This function is thread-safe and uses a mutex to synchronize
+// updates to the performance metrics.
+//
+// Parameters:
+// - duration: The time duration it took for an HTTP request to complete.
+//
+// This function should be called after each HTTP request to keep track of the
+// ConcurrencyHandler's performance over time.
+func (ch *ConcurrencyHandler) UpdatePerformanceMetrics(duration time.Duration) {
+	ch.PerfMetrics.lock.Lock()
+	defer ch.PerfMetrics.lock.Unlock()
+	ch.PerfMetrics.TotalResponseTime += duration
+	ch.PerfMetrics.TotalRequests++
+}
+
+// Min returns the smaller of the two integers.
+func Min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}