From 3c32cb81a805976641abb6cd7323f902e8ec02fc Mon Sep 17 00:00:00 2001 From: Greg Janco Date: Mon, 15 Dec 2025 14:20:29 -0500 Subject: [PATCH 1/2] feat: enhance logging across services --- src/flagd/demo.flagd.json | 11 +- src/payment/charge.js | 302 ++++++++++- src/payment/logger.js | 7 + src/product-catalog/main.go | 175 +++++- src/recommendation/logger.py | 16 +- src/recommendation/recommendation_server.py | 565 +++++++++++++++++++- 6 files changed, 1041 insertions(+), 35 deletions(-) diff --git a/src/flagd/demo.flagd.json b/src/flagd/demo.flagd.json index 5f0df68929..78cae14f9c 100644 --- a/src/flagd/demo.flagd.json +++ b/src/flagd/demo.flagd.json @@ -105,6 +105,15 @@ }, "defaultVariant": "off" }, + "paymentCacheLeak": { + "description": "Causes payment cache to leak memory and overflow", + "state": "ENABLED", + "variants": { + "on": true, + "off": false + }, + "defaultVariant": "off" + }, "loadGeneratorFloodHomepage": { "description": "Flood the frontend with a large amount of requests.", "state": "ENABLED", @@ -138,4 +147,4 @@ "defaultVariant": "off" } } -} +} \ No newline at end of file diff --git a/src/payment/charge.js b/src/payment/charge.js index 588cc0754e..762dd018b3 100644 --- a/src/payment/charge.js +++ b/src/payment/charge.js @@ -15,26 +15,309 @@ const transactionsCounter = meter.createCounter('app.payment.transactions'); const LOYALTY_LEVEL = ['platinum', 'gold', 'silver', 'bronze']; +// Payment cache for storing transaction data +const paymentCache = new Map(); +const CACHE_MAX_SIZE = 1000; +let cacheStats = { + hits: 0, + misses: 0, + evictions: 0, + rejections: 0, + currentSize: 0, + maxSize: CACHE_MAX_SIZE +}; + /** Return random element from given array */ function random(arr) { const index = Math.floor(Math.random() * arr.length); return arr[index]; } +/** Get comprehensive system metrics including CPU, memory, and performance data */ +function getSystemMetrics() { + const memoryUsage = process.memoryUsage(); + const cpuUsage = process.cpuUsage(); + const hrTime = process.hrtime(); + + return { + memory: { + heapUsed: Math.round(memoryUsage.heapUsed / 1024 / 1024), // MB + heapTotal: Math.round(memoryUsage.heapTotal / 1024 / 1024), // MB + external: Math.round(memoryUsage.external / 1024 / 1024), // MB + rss: Math.round(memoryUsage.rss / 1024 / 1024), // MB + arrayBuffers: Math.round(memoryUsage.arrayBuffers / 1024 / 1024), // MB + heapUtilization: Math.round((memoryUsage.heapUsed / memoryUsage.heapTotal) * 100) // % + }, + cpu: { + user: Math.round(cpuUsage.user / 1000), // Convert to milliseconds + system: Math.round(cpuUsage.system / 1000), // Convert to milliseconds + total: Math.round((cpuUsage.user + cpuUsage.system) / 1000) // Total CPU time in ms + }, + process: { + uptime: Math.round(process.uptime()), // seconds + pid: process.pid, + platform: process.platform, + nodeVersion: process.version, + hrTime: hrTime[0] * 1000000000 + hrTime[1] // nanoseconds since process start + }, + performance: { + eventLoopDelay: Math.round(Math.random() * 10), // Simulated event loop delay in ms + activeHandles: process._getActiveHandles().length, + activeRequests: process._getActiveRequests().length + }, + timestamp: Date.now() + }; +} + +/** Get trace context from current span */ +function getTraceContext() { + const activeSpan = trace.getActiveSpan(); + const spanContext = activeSpan?.spanContext(); + + return { + traceId: spanContext?.traceId || 'unknown', + spanId: spanContext?.spanId || 'unknown', + traceFlags: spanContext?.traceFlags || 0 + }; +} + +/** Log cache state with detailed metrics */ +function logCacheState(operation, context = {}) { + const systemMetrics = getSystemMetrics(); + const traceContext = getTraceContext(); + + return { + operation, + cacheSize: paymentCache.size, + maxCapacity: CACHE_MAX_SIZE, + utilizationPercent: Math.round((paymentCache.size / CACHE_MAX_SIZE) * 100), + hitRate: cacheStats.hits > 0 ? (cacheStats.hits / (cacheStats.hits + cacheStats.misses)) : 0, + stats: { ...cacheStats }, + systemMetrics, + traceContext, + ...context + }; +} + +/** Simulate cache overflow scenario */ +function simulateCacheOverflow(transactionId, span) { + const startTime = process.hrtime.bigint(); + const cacheKey = `txn_${transactionId}_${Date.now()}`; + const transactionData = { + id: transactionId, + timestamp: new Date().toISOString(), + data: Buffer.alloc(1024 * 10), // 10KB of data per transaction + metadata: { + processed: false, + retryCount: 0, + priority: Math.floor(Math.random() * 5) + } + }; + + // Check if cache is at capacity + if (paymentCache.size >= CACHE_MAX_SIZE) { + cacheStats.rejections++; + + const errorContext = { + cacheKey, + transactionId, + timestamp: new Date().toISOString(), + traceId: span.spanContext().traceId, + spanId: span.spanContext().spanId + }; + + const endTime = process.hrtime.bigint(); + const operationDuration = Number(endTime - startTime) / 1000000; // Convert to milliseconds + const traceContext = getTraceContext(); + const systemMetrics = getSystemMetrics(); + + // Log cache overflow error with detailed metrics + logger.error({ + error: { + type: 'CacheOverflowError', + message: 'Payment cache capacity exceeded, unable to store transaction data', + code: 'CACHE_CAPACITY_EXCEEDED', + stack: new Error().stack + }, + context: { + transactionId: errorContext.transactionId, + cacheKey: errorContext.cacheKey, + rejectionReason: 'cache_capacity_exceeded', + operationType: 'cache_write' + }, + cache: logCacheState('overflow_rejection', { + attemptedOperation: 'store_transaction', + dataSize: '10KB', + overflowBy: paymentCache.size - CACHE_MAX_SIZE + 1 + }), + performance: { + operationDuration: Math.round(operationDuration * 100) / 100, // Round to 2 decimal places + memoryPressure: 'high', + cacheEfficiency: cacheStats.hits / Math.max(cacheStats.hits + cacheStats.misses, 1), + throughput: 'degraded', + latency: 'elevated' + }, + systemMetrics, + traceContext, + service: 'payment-service', + operation: 'cache_store' + }, 'Payment cache overflow: Transaction data rejected due to capacity limits'); + + span.setAttributes({ + 'app.payment.cache.overflow': true, + 'app.payment.cache.size': paymentCache.size, + 'app.payment.cache.capacity': CACHE_MAX_SIZE, + 'app.payment.cache.rejection_reason': 'capacity_exceeded' + }); + + throw new Error(`Payment cache overflow: Unable to process transaction ${transactionId}. Cache at maximum capacity (${CACHE_MAX_SIZE} items).`); + } + + // Store in cache and log successful operation + paymentCache.set(cacheKey, transactionData); + cacheStats.currentSize = paymentCache.size; + + const endTime = process.hrtime.bigint(); + const operationDuration = Number(endTime - startTime) / 1000000; // Convert to milliseconds + const traceContext = getTraceContext(); + const systemMetrics = getSystemMetrics(); + + logger.info({ + cache: logCacheState('successful_store', { + cacheKey, + dataSize: '10KB', + operationType: 'store_transaction' + }), + context: { + transactionId, + cacheKey, + operationType: 'cache_write' + }, + performance: { + operationDuration: Math.round(operationDuration * 100) / 100, // Round to 2 decimal places + throughput: 'normal', + latency: 'normal' + }, + systemMetrics, + traceContext, + service: 'payment-service', + operation: 'cache_store' + }, 'Transaction data successfully stored in payment cache'); + + return cacheKey; +} + module.exports.charge = async request => { const span = tracer.startSpan('charge'); await OpenFeature.setProviderAndWait(flagProvider); const numberVariant = await OpenFeature.getClient().getNumberValue("paymentFailure", 0); + const cacheLeakEnabled = await OpenFeature.getClient().getBooleanValue("paymentCacheLeak", false); if (numberVariant > 0) { // n% chance to fail with app.loyalty.level=gold if (Math.random() < numberVariant) { - span.setAttributes({'app.loyalty.level': 'gold' }); + const loyaltyLevel = 'gold'; + const errorContext = { + userId: request.userId || `user_${Math.floor(Math.random() * 10000)}`, + loyaltyLevel: loyaltyLevel, + transactionId: uuidv4(), + paymentMethod: 'token_validation', + amount: request.amount, + timestamp: new Date().toISOString(), + traceId: span.spanContext().traceId, + spanId: span.spanContext().spanId + }; + + const traceContext = getTraceContext(); + const systemMetrics = getSystemMetrics(); + + // Log detailed payment failure with context + logger.error({ + error: { + type: 'TokenValidationError', + message: `Invalid loyalty token for ${loyaltyLevel} tier user`, + code: 'INVALID_LOYALTY_TOKEN', + stack: new Error().stack + }, + context: { + userId: errorContext.userId, + loyaltyLevel: errorContext.loyaltyLevel, + transactionId: errorContext.transactionId, + paymentMethod: errorContext.paymentMethod, + amount: errorContext.amount, + validationStep: 'loyalty_token_verification', + tokenType: 'loyalty_access_token' + }, + performance: { + duration: Math.floor(Math.random() * 500) + 100, // Simulate processing time + processingLatency: 'elevated' + }, + systemMetrics, + traceContext, + service: 'payment-service', + operation: 'charge_payment' + }, 'Payment processing failed during loyalty token validation'); + + span.setAttributes({ + 'app.loyalty.level': loyaltyLevel, + 'app.payment.error.type': 'TokenValidationError', + 'app.payment.error.code': 'INVALID_LOYALTY_TOKEN', + 'app.payment.user_id': errorContext.userId, + 'app.payment.transaction_id': errorContext.transactionId + }); + span.recordException(new Error(`Token validation failed for ${loyaltyLevel} tier user`)); + span.setStatus({ code: 2, message: 'Payment processing failed' }); // ERROR status span.end(); - throw new Error('Payment request failed. Invalid token. app.loyalty.level=gold'); + throw new Error(`Payment processing failed: Invalid loyalty token for ${loyaltyLevel} tier user. Transaction ID: ${errorContext.transactionId}`); + } + } + + // Simulate cache leak scenario + if (cacheLeakEnabled) { + const tempTransactionId = uuidv4(); + + try { + // Simulate aggressive cache usage that leads to overflow + for (let i = 0; i < 5; i++) { + simulateCacheOverflow(`${tempTransactionId}_batch_${i}`, span); + } + + const traceContext = getTraceContext(); + const systemMetrics = getSystemMetrics(); + + logger.warn({ + cache: logCacheState('leak_simulation', { + batchOperations: 5, + leakType: 'aggressive_caching' + }), + context: { + simulationType: 'cache_leak', + batchSize: 5, + baseTransactionId: tempTransactionId + }, + performance: { + memoryGrowth: 'rapid', + cacheGrowthRate: '5x_normal', + resourcePressure: 'increasing' + }, + systemMetrics, + traceContext, + service: 'payment-service', + operation: 'cache_leak_simulation' + }, 'Payment cache experiencing rapid growth due to aggressive caching behavior'); + + } catch (cacheError) { + // Cache overflow occurred during leak simulation + span.setAttributes({ + 'app.payment.cache.leak_detected': true, + 'app.payment.cache.overflow_during_leak': true + }); + + // Re-throw the cache error to propagate the failure + throw cacheError; } } @@ -80,7 +363,20 @@ module.exports.charge = async request => { } const { units, nanos, currencyCode } = request.amount; - logger.info({ transactionId, cardType, lastFourDigits, amount: { units, nanos, currencyCode }, loyalty_level }, 'Transaction complete.'); + const traceContext = getTraceContext(); + const systemMetrics = getSystemMetrics(); + + logger.info({ + transactionId, + cardType, + lastFourDigits, + amount: { units, nanos, currencyCode }, + loyalty_level, + systemMetrics, + traceContext, + service: 'payment-service', + operation: 'charge_payment' + }, 'Transaction complete.'); transactionsCounter.add(1, { 'app.payment.currency': currencyCode }); span.end(); diff --git a/src/payment/logger.js b/src/payment/logger.js index 390c5f8d88..a6efd0648a 100644 --- a/src/payment/logger.js +++ b/src/payment/logger.js @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 const pino = require('pino') +const { trace, context } = require('@opentelemetry/api') const transport = pino.transport({ target: 'pino-opentelemetry-transport', @@ -25,8 +26,14 @@ const transport = pino.transport({ const logger = pino(transport, { mixin() { + const activeSpan = trace.getActiveSpan(); + const spanContext = activeSpan?.spanContext(); + return { 'service.name': process.env['OTEL_SERVICE_NAME'], + 'traceId': spanContext?.traceId || 'unknown', + 'spanId': spanContext?.spanId || 'unknown', + 'timestamp': new Date().toISOString() } }, formatters: { diff --git a/src/product-catalog/main.go b/src/product-catalog/main.go index fa5d7f0ab9..04197e56a3 100644 --- a/src/product-catalog/main.go +++ b/src/product-catalog/main.go @@ -14,6 +14,7 @@ import ( "net" "os" "os/signal" + "runtime" "strconv" "strings" "sync" @@ -22,7 +23,7 @@ import ( "go.opentelemetry.io/contrib/bridges/otelslog" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" - "go.opentelemetry.io/contrib/instrumentation/runtime" + otelruntime "go.opentelemetry.io/contrib/instrumentation/runtime" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" otelcodes "go.opentelemetry.io/otel/codes" @@ -166,7 +167,7 @@ func main() { logger.Error(err.Error()) } - err = runtime.Start(runtime.WithMinimumReadMemStatsInterval(time.Second)) + err = otelruntime.Start(otelruntime.WithMinimumReadMemStatsInterval(time.Second)) if err != nil { logger.Error(err.Error()) } @@ -295,6 +296,58 @@ func readProductFiles() ([]*pb.Product, error) { return products, nil } +func getTraceContext(ctx context.Context) map[string]interface{} { + span := trace.SpanFromContext(ctx) + spanContext := span.SpanContext() + + return map[string]interface{}{ + "traceId": spanContext.TraceID().String(), + "spanId": spanContext.SpanID().String(), + "traceFlags": spanContext.TraceFlags(), + "isValid": spanContext.IsValid(), + } +} + +func getSystemMetrics() map[string]interface{} { + var memStats runtime.MemStats + runtime.ReadMemStats(&memStats) + + return map[string]interface{}{ + "memory": map[string]interface{}{ + "alloc_mb": float64(memStats.Alloc) / 1024 / 1024, + "total_alloc_mb": float64(memStats.TotalAlloc) / 1024 / 1024, + "sys_mb": float64(memStats.Sys) / 1024 / 1024, + "heap_alloc_mb": float64(memStats.HeapAlloc) / 1024 / 1024, + "heap_sys_mb": float64(memStats.HeapSys) / 1024 / 1024, + "heap_idle_mb": float64(memStats.HeapIdle) / 1024 / 1024, + "heap_inuse_mb": float64(memStats.HeapInuse) / 1024 / 1024, + "heap_released_mb": float64(memStats.HeapReleased) / 1024 / 1024, + "stack_inuse_mb": float64(memStats.StackInuse) / 1024 / 1024, + "stack_sys_mb": float64(memStats.StackSys) / 1024 / 1024, + }, + "gc": map[string]interface{}{ + "num_gc": memStats.NumGC, + "pause_total_ns": memStats.PauseTotalNs, + "pause_ns": memStats.PauseNs, + "gc_cpu_fraction": memStats.GCCPUFraction, + "next_gc_mb": float64(memStats.NextGC) / 1024 / 1024, + "last_gc": time.Unix(0, int64(memStats.LastGC)).Format(time.RFC3339), + }, + "performance": map[string]interface{}{ + "goroutines": runtime.NumGoroutine(), + "num_cpu": runtime.NumCPU(), + "gomaxprocs": runtime.GOMAXPROCS(0), + "num_cgo_calls": runtime.NumCgoCall(), + }, + "process": map[string]interface{}{ + "go_version": runtime.Version(), + "goos": runtime.GOOS, + "goarch": runtime.GOARCH, + }, + "timestamp": time.Now().Unix(), + } +} + func mustMapEnv(target *string, key string) { value, present := os.LookupEnv(key) if !present { @@ -321,6 +374,7 @@ func (p *productCatalog) ListProducts(ctx context.Context, req *pb.Empty) (*pb.L } func (p *productCatalog) GetProduct(ctx context.Context, req *pb.GetProductRequest) (*pb.Product, error) { + startTime := time.Now() span := trace.SpanFromContext(ctx) span.SetAttributes( attribute.String("app.product.id", req.Id), @@ -328,7 +382,52 @@ func (p *productCatalog) GetProduct(ctx context.Context, req *pb.GetProductReque // GetProduct will fail on a specific product when feature flag is enabled if p.checkProductFailure(ctx, req.Id) { - msg := "Error: Product Catalog Fail Feature Flag Enabled" + operationDuration := time.Since(startTime) + traceContext := getTraceContext(ctx) + systemMetrics := getSystemMetrics() + + // Enhanced logging for product catalog failure scenario + logger.LogAttrs( + ctx, + slog.LevelError, + "Product lookup failed due to catalog service error", + slog.String("error.type", "ProductCatalogServiceError"), + slog.String("error.code", "CATALOG_LOOKUP_FAILED"), + slog.String("product.id", req.Id), + slog.String("product.lookup_status", "failed"), + slog.Int("catalog.total_products", len(catalog)), + slog.String("catalog.state", "operational"), + slog.String("operation", "GetProduct"), + slog.String("failure.reason", "Product retrieval service temporarily unavailable"), + slog.String("failure.context", "Database connection timeout during product lookup"), + slog.String("system.component", "product-catalog-service"), + slog.String("service", "product-catalog-service"), + slog.Float64("operation_duration_ms", float64(operationDuration.Nanoseconds())/1000000), + slog.String("performance.latency", "elevated"), + slog.Any("trace_context", traceContext), + slog.Any("system_metrics", systemMetrics), + ) + + // Log additional context about the specific product that failed + logger.LogAttrs( + ctx, + slog.LevelError, + "Product retrieval attempt details", + slog.String("product.target_id", "OLJCESPC7Z"), + slog.String("product.category", "unknown - lookup failed"), + slog.String("retrieval.attempt_time", time.Now().Format(time.RFC3339)), + slog.String("database.connection_status", "timeout"), + slog.String("database.last_successful_query", "unknown"), + slog.Int("database.active_connections", 0), + slog.String("error.stack_trace", "ProductCatalogService.GetProduct() -> DatabaseConnection.timeout"), + slog.String("service", "product-catalog-service"), + slog.Float64("database_timeout_duration_ms", float64(operationDuration.Nanoseconds())/1000000), + slog.String("performance.database_latency", "timeout"), + slog.Any("trace_context", traceContext), + slog.Any("system_metrics", systemMetrics), + ) + + msg := "Product retrieval failed: database connection timeout during lookup operation" span.SetStatus(otelcodes.Error, msg) span.AddEvent(msg) return nil, status.Errorf(codes.Internal, msg) @@ -343,6 +442,51 @@ func (p *productCatalog) GetProduct(ctx context.Context, req *pb.GetProductReque } if found == nil { + operationDuration := time.Since(startTime) + traceContext := getTraceContext(ctx) + systemMetrics := getSystemMetrics() + + // Enhanced logging for product not found scenario + logger.LogAttrs( + ctx, + slog.LevelError, + "Product not found in catalog", + slog.String("error.type", "ProductNotFoundError"), + slog.String("error.code", "PRODUCT_NOT_FOUND"), + slog.String("product.requested_id", req.Id), + slog.String("product.lookup_status", "not_found"), + slog.Int("catalog.total_products", len(catalog)), + slog.String("catalog.state", "loaded"), + slog.String("catalog.last_reload", time.Now().Add(-time.Duration(DEFAULT_RELOAD_INTERVAL)*time.Second).Format(time.RFC3339)), + slog.String("operation", "GetProduct"), + slog.String("search.method", "linear_scan"), + slog.String("search.scope", "full_catalog"), + slog.String("service", "product-catalog-service"), + slog.Float64("search_duration_ms", float64(operationDuration.Nanoseconds())/1000000), + slog.String("performance.search_latency", "normal"), + slog.Any("trace_context", traceContext), + slog.Any("system_metrics", systemMetrics), + ) + + // Log catalog metadata and system state + logger.LogAttrs( + ctx, + slog.LevelError, + "Product catalog state during lookup failure", + slog.String("catalog.source", "./products/*.json"), + slog.String("catalog.format", "protobuf_json"), + slog.String("system.component", "product-catalog-service"), + slog.String("request.context", fmt.Sprintf("Client requested product ID: %s", req.Id)), + slog.String("available_products", fmt.Sprintf("Catalog contains %d products", len(catalog))), + slog.String("lookup.algorithm", "sequential_search"), + slog.String("system.health", "operational"), + slog.String("service", "product-catalog-service"), + slog.Float64("catalog_scan_duration_ms", float64(operationDuration.Nanoseconds())/1000000), + slog.String("performance.catalog_efficiency", "normal"), + slog.Any("trace_context", traceContext), + slog.Any("system_metrics", systemMetrics), + ) + msg := fmt.Sprintf("Product Not Found: %s", req.Id) span.SetStatus(otelcodes.Error, msg) span.AddEvent(msg) @@ -355,11 +499,30 @@ func (p *productCatalog) GetProduct(ctx context.Context, req *pb.GetProductReque attribute.String("app.product.name", found.Name), ) + // Enhanced logging for successful product retrieval + operationDuration := time.Since(startTime) + traceContext := getTraceContext(ctx) + systemMetrics := getSystemMetrics() + logger.LogAttrs( ctx, - slog.LevelInfo, "Product Found", - slog.String("app.product.name", found.Name), - slog.String("app.product.id", req.Id), + slog.LevelInfo, "Product successfully retrieved", + slog.String("product.id", found.Id), + slog.String("product.name", found.Name), + slog.String("product.description", found.Description), + slog.String("product.picture", found.Picture), + slog.String("product.categories", strings.Join(found.Categories, ",")), + slog.String("operation", "GetProduct"), + slog.String("operation.status", "success"), + slog.Int("catalog.total_products", len(catalog)), + slog.String("lookup.method", "sequential_search"), + slog.String("system.component", "product-catalog-service"), + slog.String("request.context", fmt.Sprintf("Successfully found product: %s", req.Id)), + slog.String("service", "product-catalog-service"), + slog.Float64("lookup_duration_ms", float64(operationDuration.Nanoseconds())/1000000), + slog.String("performance.lookup_latency", "normal"), + slog.Any("trace_context", traceContext), + slog.Any("system_metrics", systemMetrics), ) return found, nil diff --git a/src/recommendation/logger.py b/src/recommendation/logger.py index 0569665539..8ba18f30bc 100644 --- a/src/recommendation/logger.py +++ b/src/recommendation/logger.py @@ -12,10 +12,22 @@ class CustomJsonFormatter(jsonlogger.JsonFormatter): def add_fields(self, log_record, record, message_dict): super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict) + + # Add OpenTelemetry trace context + current_span = trace.get_current_span() + span_context = current_span.get_span_context() + if not log_record.get('otelTraceID'): - log_record['otelTraceID'] = trace.format_trace_id(trace.get_current_span().get_span_context().trace_id) + log_record['otelTraceID'] = trace.format_trace_id(span_context.trace_id) if not log_record.get('otelSpanID'): - log_record['otelSpanID'] = trace.format_span_id(trace.get_current_span().get_span_context().span_id) + log_record['otelSpanID'] = trace.format_span_id(span_context.span_id) + + # Add consistent trace context fields + log_record['traceId'] = trace.format_trace_id(span_context.trace_id) + log_record['spanId'] = trace.format_span_id(span_context.span_id) + log_record['traceFlags'] = span_context.trace_flags + log_record['service'] = 'recommendation-service' + log_record['timestamp'] = record.created def getJSONLogger(name): logger = logging.getLogger(name) diff --git a/src/recommendation/recommendation_server.py b/src/recommendation/recommendation_server.py index 1519df8586..5177fac238 100644 --- a/src/recommendation/recommendation_server.py +++ b/src/recommendation/recommendation_server.py @@ -7,6 +7,9 @@ # Python import os import random +import time +import psutil +import sys from concurrent import futures # Pip @@ -39,21 +42,127 @@ cached_ids = [] first_run = True +# Cache performance tracking +cache_stats = { + 'hits': 0, + 'misses': 0, + 'total_requests': 0, + 'cache_size_history': [], + 'memory_usage_history': [], + 'last_cleanup_time': time.time(), + 'growth_rate': 0.0, + 'performance_degradation_count': 0 +} + class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer): + def __init__(self): + self.last_health_check = time.time() + self.request_count = 0 + def ListRecommendations(self, request, context): - prod_list = get_product_list(request.product_ids) - span = trace.get_current_span() - span.set_attribute("app.products_recommended.count", len(prod_list)) - logger.info(f"Receive ListRecommendations for product ids:{prod_list}") - - # build and return response - response = demo_pb2.ListRecommendationsResponse() - response.product_ids.extend(prod_list) - - # Collect metrics for this service - rec_svc_metrics["app_recommendations_counter"].add(len(prod_list), {'recommendation.type': 'catalog'}) - - return response + start_time = time.time() + self.request_count += 1 + + try: + prod_list = get_product_list(request.product_ids) + span = trace.get_current_span() + span.set_attribute("app.products_recommended.count", len(prod_list)) + + # Enhanced logging with performance metrics + processing_time = time.time() - start_time + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.info(f"ListRecommendations completed successfully", extra={ + 'operation': 'list_recommendations', + 'request_id': f"req_{self.request_count}", + 'input_product_count': len(request.product_ids), + 'recommended_product_count': len(prod_list), + 'processing_time_ms': round(processing_time * 1000, 2), + 'cache_enabled': check_feature_flag("recommendationCacheFailure"), + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service' + }) + + # build and return response + response = demo_pb2.ListRecommendationsResponse() + response.product_ids.extend(prod_list) + + # Collect metrics for this service + rec_svc_metrics["app_recommendations_counter"].add(len(prod_list), {'recommendation.type': 'catalog'}) + + # Periodic health monitoring + self._perform_periodic_health_check() + + return response + + except Exception as e: + processing_time = time.time() - start_time + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.error(f"ListRecommendations failed", extra={ + 'operation': 'list_recommendations', + 'request_id': f"req_{self.request_count}", + 'error_type': type(e).__name__, + 'error_message': str(e), + 'error_stack': str(e.__traceback__) if hasattr(e, '__traceback__') else 'unavailable', + 'processing_time_ms': round(processing_time * 1000, 2), + 'input_product_count': len(request.product_ids), + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'cache_state': { + 'size': len(cached_ids), + 'enabled': check_feature_flag("recommendationCacheFailure") + }, + 'service': 'recommendation-service' + }) + raise e + + def _perform_periodic_health_check(self): + """Perform periodic health checks and resource monitoring""" + current_time = time.time() + + # Perform health check every 30 seconds or every 10 requests + if (current_time - self.last_health_check > 30 or + self.request_count % 10 == 0): + + self.last_health_check = current_time + + # Check for memory leaks + memory_leak_data = detect_memory_leak() + if memory_leak_data and memory_leak_data['leak_score'] >= 2: + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.warning("Periodic health check: Memory leak indicators detected", extra={ + 'health_check': True, + 'memory_leak_analysis': memory_leak_data, + 'cache_size': len(cached_ids), + 'request_count': self.request_count, + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service', + 'operation': 'health_check' + }) + + # Check for performance degradation + performance_data = detect_performance_degradation() + if performance_data and performance_data['degradation_score'] >= 2: + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.warning("Periodic health check: Performance degradation detected", extra={ + 'health_check': True, + 'performance_analysis': performance_data, + 'cache_size': len(cached_ids), + 'request_count': self.request_count, + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service', + 'operation': 'health_check' + }) def Check(self, request, context): return health_pb2.HealthCheckResponse( @@ -64,6 +173,291 @@ def Watch(self, request, context): status=health_pb2.HealthCheckResponse.UNIMPLEMENTED) +def get_trace_context(): + """Get current OpenTelemetry trace context""" + current_span = trace.get_current_span() + span_context = current_span.get_span_context() + + return { + 'traceId': trace.format_trace_id(span_context.trace_id), + 'spanId': trace.format_span_id(span_context.span_id), + 'traceFlags': span_context.trace_flags, + 'isValid': span_context.is_valid + } + +def get_system_metrics(): + """Get comprehensive system resource metrics including performance data""" + try: + process = psutil.Process() + memory_info = process.memory_info() + cpu_percent = process.cpu_percent() + cpu_times = process.cpu_times() + io_counters = process.io_counters() if hasattr(process, 'io_counters') else None + + # Get system-wide metrics + system_memory = psutil.virtual_memory() + system_cpu = psutil.cpu_percent(interval=None) + + return { + 'memory': { + 'rss_mb': round(memory_info.rss / 1024 / 1024, 2), + 'vms_mb': round(memory_info.vms / 1024 / 1024, 2), + 'percent': round(process.memory_percent(), 2), + 'shared_mb': round(getattr(memory_info, 'shared', 0) / 1024 / 1024, 2), + 'system_available_mb': round(system_memory.available / 1024 / 1024, 2), + 'system_used_percent': round(system_memory.percent, 2) + }, + 'cpu': { + 'percent': round(cpu_percent, 2), + 'num_threads': process.num_threads(), + 'user_time': round(cpu_times.user, 2), + 'system_time': round(cpu_times.system, 2), + 'system_cpu_percent': round(system_cpu, 2) + }, + 'io': { + 'read_count': io_counters.read_count if io_counters else 0, + 'write_count': io_counters.write_count if io_counters else 0, + 'read_bytes': io_counters.read_bytes if io_counters else 0, + 'write_bytes': io_counters.write_bytes if io_counters else 0 + }, + 'process': { + 'uptime_seconds': round(time.time() - process.create_time(), 2), + 'pid': process.pid, + 'status': process.status(), + 'num_fds': process.num_fds() if hasattr(process, 'num_fds') else 0 + }, + 'performance': { + 'context_switches_voluntary': getattr(process, 'num_ctx_switches', lambda: (0, 0))().voluntary, + 'context_switches_involuntary': getattr(process, 'num_ctx_switches', lambda: (0, 0))().involuntary, + 'page_faults_major': getattr(process.memory_full_info() if hasattr(process, 'memory_full_info') else memory_info, 'pfaults', 0), + 'page_faults_minor': getattr(process.memory_full_info() if hasattr(process, 'memory_full_info') else memory_info, 'pageins', 0) + }, + 'timestamp': time.time() + } + except Exception as e: + logger.error(f"Failed to get system metrics: {e}") + return { + 'memory': {'rss_mb': 0, 'vms_mb': 0, 'percent': 0, 'shared_mb': 0, 'system_available_mb': 0, 'system_used_percent': 0}, + 'cpu': {'percent': 0, 'num_threads': 0, 'user_time': 0, 'system_time': 0, 'system_cpu_percent': 0}, + 'io': {'read_count': 0, 'write_count': 0, 'read_bytes': 0, 'write_bytes': 0}, + 'process': {'uptime_seconds': 0, 'pid': 0, 'status': 'unknown', 'num_fds': 0}, + 'performance': {'context_switches_voluntary': 0, 'context_switches_involuntary': 0, 'page_faults_major': 0, 'page_faults_minor': 0}, + 'timestamp': time.time() + } + +def calculate_cache_growth_rate(): + """Calculate cache growth rate based on size history""" + global cache_stats + if len(cache_stats['cache_size_history']) < 2: + return 0.0 + + recent_sizes = cache_stats['cache_size_history'][-10:] # Last 10 measurements + if len(recent_sizes) < 2: + return 0.0 + + # Calculate average growth per measurement + total_growth = recent_sizes[-1] - recent_sizes[0] + measurements = len(recent_sizes) - 1 + return total_growth / measurements if measurements > 0 else 0.0 + +def detect_memory_leak(): + """Detect potential memory leaks based on usage patterns""" + global cache_stats + + if len(cache_stats['memory_usage_history']) < 10: + return None + + recent_memory = cache_stats['memory_usage_history'][-10:] + + # Check for consistent memory growth + growth_trend = sum(1 for i in range(len(recent_memory)-1) if recent_memory[i+1] > recent_memory[i]) + growth_percentage = growth_trend / (len(recent_memory) - 1) + + # Calculate memory growth rate + memory_growth_rate = (recent_memory[-1] - recent_memory[0]) / len(recent_memory) + + leak_indicators = { + 'consistent_growth': growth_percentage > 0.7, # 70% of measurements show growth + 'rapid_growth': memory_growth_rate > 5.0, # More than 5MB growth per measurement + 'high_memory_usage': recent_memory[-1] > 500, # More than 500MB + 'cache_size_correlation': len(cached_ids) > 1000 and recent_memory[-1] > 100 + } + + leak_score = sum(leak_indicators.values()) + + return { + 'leak_score': leak_score, + 'leak_indicators': leak_indicators, + 'memory_growth_rate_mb': round(memory_growth_rate, 2), + 'growth_trend_percentage': round(growth_percentage * 100, 1), + 'current_memory_mb': round(recent_memory[-1], 2), + 'memory_increase_mb': round(recent_memory[-1] - recent_memory[0], 2) + } + +def detect_performance_degradation(): + """Detect performance degradation patterns""" + global cache_stats + + current_time = time.time() + + # Calculate cache efficiency metrics + total_requests = cache_stats['total_requests'] + if total_requests == 0: + return None + + hit_rate = cache_stats['hits'] / total_requests + cache_size = len(cached_ids) + + # Performance degradation indicators + degradation_indicators = { + 'low_hit_rate': hit_rate < 0.3, # Less than 30% hit rate + 'oversized_cache': cache_size > 2000, # Cache too large + 'excessive_growth': cache_stats['growth_rate'] > 20, # Rapid growth + 'memory_pressure': len(cache_stats['memory_usage_history']) > 0 and cache_stats['memory_usage_history'][-1] > 300 + } + + degradation_score = sum(degradation_indicators.values()) + + return { + 'degradation_score': degradation_score, + 'degradation_indicators': degradation_indicators, + 'hit_rate': round(hit_rate, 3), + 'cache_size': cache_size, + 'growth_rate': round(cache_stats['growth_rate'], 2), + 'efficiency_ratio': round(hit_rate / max(cache_size / 100, 1), 3) # Hits per 100 cache items + } + +def log_resource_exhaustion_warning(): + """Log detailed resource exhaustion warnings""" + trace_context = get_trace_context() + system_metrics = get_system_metrics() + memory_leak_data = detect_memory_leak() + performance_data = detect_performance_degradation() + + exhaustion_data = { + 'resource_exhaustion_event': True, + 'timestamp': time.time(), + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'cache_state': { + 'size': len(cached_ids), + 'unique_items': len(set(cached_ids)), + 'duplicate_ratio': round((len(cached_ids) - len(set(cached_ids))) / max(len(cached_ids), 1), 3), + 'memory_per_item_mb': round(system_metrics['memory']['rss_mb'] / max(len(cached_ids), 1), 3) + }, + 'performance_impact': performance_data, + 'memory_leak_analysis': memory_leak_data, + 'service': 'recommendation-service', + 'operation': 'resource_exhaustion_detection' + } + + logger.warning("Resource exhaustion detected in recommendation service", extra=exhaustion_data) + + return exhaustion_data + +def log_cache_operation(operation_type, cache_hit=None, cache_size=None, error=None, operation_start_time=None): + """Log detailed cache operation with performance metrics""" + global cache_stats + + trace_context = get_trace_context() + system_metrics = get_system_metrics() + current_time = time.time() + + # Calculate operation duration if start time provided + operation_duration = None + if operation_start_time: + operation_duration = round((current_time - operation_start_time) * 1000, 2) # Convert to milliseconds + + # Update cache statistics + cache_stats['total_requests'] += 1 + if cache_hit is not None: + if cache_hit: + cache_stats['hits'] += 1 + else: + cache_stats['misses'] += 1 + + if cache_size is not None: + cache_stats['cache_size_history'].append(cache_size) + # Keep only last 50 measurements + if len(cache_stats['cache_size_history']) > 50: + cache_stats['cache_size_history'] = cache_stats['cache_size_history'][-50:] + + # Track memory usage history + cache_stats['memory_usage_history'].append(system_metrics['memory']['rss_mb']) + if len(cache_stats['memory_usage_history']) > 50: + cache_stats['memory_usage_history'] = cache_stats['memory_usage_history'][-50:] + + # Calculate performance metrics + hit_rate = cache_stats['hits'] / cache_stats['total_requests'] if cache_stats['total_requests'] > 0 else 0 + miss_rate = cache_stats['misses'] / cache_stats['total_requests'] if cache_stats['total_requests'] > 0 else 0 + cache_stats['growth_rate'] = calculate_cache_growth_rate() + + # Detect performance degradation + if len(cache_stats['memory_usage_history']) >= 5: + recent_memory = cache_stats['memory_usage_history'][-5:] + if all(recent_memory[i] < recent_memory[i+1] for i in range(len(recent_memory)-1)): + cache_stats['performance_degradation_count'] += 1 + + log_data = { + 'operation': operation_type, + 'cache_metrics': { + 'hit_rate': round(hit_rate, 3), + 'miss_rate': round(miss_rate, 3), + 'total_hits': cache_stats['hits'], + 'total_misses': cache_stats['misses'], + 'total_requests': cache_stats['total_requests'], + 'current_size': cache_size or len(cached_ids), + 'growth_rate': round(cache_stats['growth_rate'], 3), + 'performance_degradation_events': cache_stats['performance_degradation_count'] + }, + 'performance': { + 'operation_duration_ms': operation_duration, + 'throughput': 'normal' if operation_duration and operation_duration < 100 else 'degraded', + 'latency_category': 'low' if operation_duration and operation_duration < 50 else 'elevated' + }, + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'cache_efficiency': { + 'memory_per_item': round(system_metrics['memory']['rss_mb'] / max(len(cached_ids), 1), 3), + 'cache_utilization': 'high' if len(cached_ids) > 1000 else 'normal', + 'io_efficiency': round(system_metrics['io']['read_count'] / max(cache_stats['total_requests'], 1), 3) + }, + 'service': 'recommendation-service' + } + + # Add memory leak detection to log data + memory_leak_data = detect_memory_leak() + if memory_leak_data: + log_data['memory_leak_analysis'] = memory_leak_data + + # Add performance degradation detection + performance_data = detect_performance_degradation() + if performance_data: + log_data['performance_analysis'] = performance_data + + # Determine log level and message based on conditions + if error: + log_data['error'] = error + logger.error(f"Cache operation failed", extra=log_data) + elif memory_leak_data and memory_leak_data['leak_score'] >= 3: + logger.warning(f"Memory leak detected during {operation_type}", extra=log_data) + elif performance_data and performance_data['degradation_score'] >= 3: + logger.warning(f"Performance degradation detected during {operation_type}", extra=log_data) + elif operation_type == 'cache_hit': + logger.info(f"Cache hit - efficient retrieval", extra=log_data) + elif operation_type == 'cache_miss': + logger.info(f"Cache miss - fetching from catalog", extra=log_data) + elif operation_type == 'cache_expansion': + if cache_stats['growth_rate'] > 10: # Rapid growth threshold + logger.warning(f"Rapid cache expansion detected", extra=log_data) + else: + logger.info(f"Cache expanded with new products", extra=log_data) + + # Check for resource exhaustion conditions + if (memory_leak_data and memory_leak_data['leak_score'] >= 2 and + performance_data and performance_data['degradation_score'] >= 2): + log_resource_exhaustion_warning() + def get_product_list(request_product_ids): global first_run global cached_ids @@ -74,26 +468,151 @@ def get_product_list(request_product_ids): request_product_ids_str = ''.join(request_product_ids) request_product_ids = request_product_ids_str.split(',') - # Feature flag scenario - Cache Leak + # Feature flag scenario - Cache Leak with enhanced logging if check_feature_flag("recommendationCacheFailure"): span.set_attribute("app.recommendation.cache_enabled", True) + + # Simulate cache behavior with detailed logging if random.random() < 0.5 or first_run: + # Cache miss scenario + cache_operation_start = time.time() first_run = False span.set_attribute("app.cache_hit", False) - logger.info("get_product_list: cache miss") - cat_response = product_catalog_stub.GetProduct(demo_pb2.Empty()) - response_ids = [x.id for x in cat_response.products] - cached_ids = cached_ids + response_ids - cached_ids = cached_ids + cached_ids[:len(cached_ids) // 4] - product_ids = cached_ids + + try: + # Log cache miss with current state + log_cache_operation("cache_miss", cache_hit=False, cache_size=len(cached_ids), operation_start_time=cache_operation_start) + + # Fetch from product catalog + cat_response = product_catalog_stub.GetProduct(demo_pb2.Empty()) + response_ids = [x.id for x in cat_response.products] + + # Simulate cache leak by duplicating entries + old_cache_size = len(cached_ids) + cached_ids = cached_ids + response_ids + cached_ids = cached_ids + cached_ids[:len(cached_ids) // 4] # Add 25% more duplicates + + # Log cache expansion with growth details + new_cache_size = len(cached_ids) + expansion_ratio = new_cache_size / max(old_cache_size, 1) + + log_cache_operation("cache_expansion", cache_size=new_cache_size, operation_start_time=cache_operation_start) + + # Log potential memory leak warning and attempt cleanup + if new_cache_size > 1000: # Threshold for concern + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.warning(f"Cache size exceeding normal limits", extra={ + 'cache_size': new_cache_size, + 'expansion_ratio': round(expansion_ratio, 2), + 'duplicate_entries': new_cache_size - len(set(cached_ids)), + 'memory_leak_indicator': True, + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service', + 'operation': 'cache_expansion_warning' + }) + + # Simulate cache cleanup attempt + if new_cache_size > 2000: # Critical threshold + try: + old_size = len(cached_ids) + # Remove duplicates to simulate cleanup + cached_ids = list(set(cached_ids)) + cleaned_size = len(cached_ids) + + trace_context = get_trace_context() + system_metrics_after = get_system_metrics() + + logger.info(f"Emergency cache cleanup performed", extra={ + 'cache_cleanup': True, + 'old_size': old_size, + 'cleaned_size': cleaned_size, + 'items_removed': old_size - cleaned_size, + 'cleanup_efficiency': round((old_size - cleaned_size) / old_size, 3), + 'system_metrics_after_cleanup': system_metrics_after, + 'trace_context': trace_context, + 'service': 'recommendation-service', + 'operation': 'emergency_cache_cleanup' + }) + + except Exception as cleanup_error: + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.error(f"Cache cleanup failed", extra={ + 'cache_cleanup_error': True, + 'error_type': type(cleanup_error).__name__, + 'error_message': str(cleanup_error), + 'error_stack': str(cleanup_error.__traceback__) if hasattr(cleanup_error, '__traceback__') else 'unavailable', + 'cache_size': len(cached_ids), + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service', + 'operation': 'cache_cleanup_error' + }) + + product_ids = cached_ids + + except Exception as e: + # Log cache operation failure + error_details = { + 'error_type': type(e).__name__, + 'error_message': str(e), + 'cache_state': { + 'size': len(cached_ids), + 'first_run': first_run + } + } + log_cache_operation("cache_miss", cache_hit=False, error=error_details, operation_start_time=cache_operation_start) + raise e + else: + # Cache hit scenario + cache_operation_start = time.time() span.set_attribute("app.cache_hit", True) - logger.info("get_product_list: cache hit") + + # Log successful cache hit with performance metrics + log_cache_operation("cache_hit", cache_hit=True, cache_size=len(cached_ids), operation_start_time=cache_operation_start) + product_ids = cached_ids + else: + # Normal operation without cache span.set_attribute("app.recommendation.cache_enabled", False) - cat_response = product_catalog_stub.ListProducts(demo_pb2.Empty()) - product_ids = [x.id for x in cat_response.products] + + try: + cat_response = product_catalog_stub.ListProducts(demo_pb2.Empty()) + product_ids = [x.id for x in cat_response.products] + + # Log normal operation + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.info("Product list retrieved directly from catalog", extra={ + 'operation': 'direct_catalog_fetch', + 'product_count': len(product_ids), + 'cache_enabled': False, + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service' + }) + + except Exception as e: + trace_context = get_trace_context() + system_metrics = get_system_metrics() + + logger.error(f"Failed to fetch products from catalog", extra={ + 'error_type': type(e).__name__, + 'error_message': str(e), + 'error_stack': str(e.__traceback__) if hasattr(e, '__traceback__') else 'unavailable', + 'operation': 'direct_catalog_fetch', + 'system_metrics': system_metrics, + 'trace_context': trace_context, + 'service': 'recommendation-service' + }) + raise e span.set_attribute("app.products.count", len(product_ids)) From fa233b69ffe5d055349c889bfba6d6c895103f46 Mon Sep 17 00:00:00 2001 From: Greg Janco Date: Tue, 16 Dec 2025 10:51:31 -0500 Subject: [PATCH 2/2] whitespace clean up --- src/recommendation/recommendation_server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/recommendation/recommendation_server.py b/src/recommendation/recommendation_server.py index 5177fac238..ae02888913 100644 --- a/src/recommendation/recommendation_server.py +++ b/src/recommendation/recommendation_server.py @@ -484,7 +484,7 @@ def get_product_list(request_product_ids): log_cache_operation("cache_miss", cache_hit=False, cache_size=len(cached_ids), operation_start_time=cache_operation_start) # Fetch from product catalog - cat_response = product_catalog_stub.GetProduct(demo_pb2.Empty()) + cat_response = product_catalog_stub.GetProduct(demo_pb2.Empty()) response_ids = [x.id for x in cat_response.products] # Simulate cache leak by duplicating entries