timeouts for faas
This commit is contained in:
@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
@ -35,7 +36,7 @@ func NewExecutionService(
|
||||
}
|
||||
|
||||
func (s *executionService) Execute(ctx context.Context, req *domain.ExecuteFunctionRequest, userID string) (*domain.ExecuteFunctionResponse, error) {
|
||||
s.logger.Info("Executing function",
|
||||
s.logger.Info("🔄 TIMEOUT DEBUG: Starting function execution",
|
||||
zap.String("function_id", req.FunctionID.String()),
|
||||
zap.String("user_id", userID),
|
||||
zap.Bool("async", req.Async))
|
||||
@ -45,6 +46,11 @@ func (s *executionService) Execute(ctx context.Context, req *domain.ExecuteFunct
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("function not found: %w", err)
|
||||
}
|
||||
|
||||
s.logger.Info("⏱️ TIMEOUT DEBUG: Function loaded with timeout",
|
||||
zap.String("function_id", req.FunctionID.String()),
|
||||
zap.String("function_name", function.Name),
|
||||
zap.Duration("timeout", function.Timeout.Duration))
|
||||
|
||||
// Create execution record
|
||||
// Initialize input with empty JSON if nil or empty
|
||||
@ -108,15 +114,52 @@ func (s *executionService) executeSync(ctx context.Context, execution *domain.Fu
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Create timeout context for execution
|
||||
execCtx := ctx
|
||||
var cancel context.CancelFunc
|
||||
if function.Timeout.Duration > 0 {
|
||||
s.logger.Info("⏰ TIMEOUT DEBUG: Creating execution timeout context for sync execution",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()),
|
||||
zap.Duration("timeout", function.Timeout.Duration))
|
||||
execCtx, cancel = context.WithTimeout(ctx, function.Timeout.Duration)
|
||||
defer cancel()
|
||||
} else {
|
||||
s.logger.Info("❌ TIMEOUT DEBUG: No timeout specified for sync execution",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()))
|
||||
}
|
||||
|
||||
// Execute function
|
||||
result, err := backend.Execute(ctx, function, execution.Input)
|
||||
s.logger.Debug("Starting function execution in runtime backend (sync)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()),
|
||||
zap.String("runtime", string(function.Runtime)))
|
||||
result, err := backend.Execute(execCtx, function, execution.Input)
|
||||
s.logger.Debug("Function execution completed in runtime backend (sync)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()),
|
||||
zap.Error(err))
|
||||
if err != nil {
|
||||
execution.Status = domain.StatusFailed
|
||||
execution.Error = fmt.Sprintf("execution failed: %v", err)
|
||||
// Check if this was a timeout error
|
||||
if execCtx.Err() == context.DeadlineExceeded {
|
||||
s.logger.Debug("Sync execution timed out (context deadline exceeded)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.Duration("timeout", function.Timeout.Duration),
|
||||
zap.Error(execCtx.Err()))
|
||||
execution.Status = domain.StatusTimeout
|
||||
execution.Error = fmt.Sprintf("function execution timed out after %v", function.Timeout.Duration)
|
||||
} else {
|
||||
s.logger.Debug("Sync execution failed (non-timeout error)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.Error(err))
|
||||
execution.Status = domain.StatusFailed
|
||||
execution.Error = fmt.Sprintf("execution failed: %v", err)
|
||||
}
|
||||
s.updateExecutionComplete(ctx, execution)
|
||||
return &domain.ExecuteFunctionResponse{
|
||||
ExecutionID: execution.ID,
|
||||
Status: domain.StatusFailed,
|
||||
Status: execution.Status,
|
||||
Error: execution.Error,
|
||||
}, nil
|
||||
}
|
||||
@ -132,12 +175,30 @@ func (s *executionService) executeSync(ctx context.Context, execution *domain.Fu
|
||||
execution.Error = result.Error
|
||||
execution.Duration = result.Duration
|
||||
execution.MemoryUsed = result.MemoryUsed
|
||||
s.updateExecutionComplete(ctx, execution)
|
||||
|
||||
// Check if the result indicates a timeout
|
||||
if result.Error != "" {
|
||||
execution.Status = domain.StatusFailed
|
||||
s.logger.Debug("Execution result contains error",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("error", result.Error))
|
||||
if strings.Contains(result.Error, "timed out") {
|
||||
s.logger.Debug("Result indicates timeout - setting status to timeout",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("error", result.Error))
|
||||
execution.Status = domain.StatusTimeout
|
||||
} else {
|
||||
s.logger.Debug("Result indicates failure - setting status to failed",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("error", result.Error))
|
||||
execution.Status = domain.StatusFailed
|
||||
}
|
||||
} else {
|
||||
s.logger.Debug("Execution successful - no errors",
|
||||
zap.String("execution_id", execution.ID.String()))
|
||||
}
|
||||
|
||||
s.updateExecutionComplete(ctx, execution)
|
||||
|
||||
return &domain.ExecuteFunctionResponse{
|
||||
ExecutionID: execution.ID,
|
||||
Status: execution.Status,
|
||||
@ -169,14 +230,51 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
|
||||
return
|
||||
}
|
||||
|
||||
// Create timeout context for execution
|
||||
execCtx := ctx
|
||||
var cancel context.CancelFunc
|
||||
if function.Timeout.Duration > 0 {
|
||||
s.logger.Debug("Creating execution timeout context for async execution",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()),
|
||||
zap.Duration("timeout", function.Timeout.Duration))
|
||||
execCtx, cancel = context.WithTimeout(ctx, function.Timeout.Duration)
|
||||
defer cancel()
|
||||
} else {
|
||||
s.logger.Debug("No timeout specified for async execution",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()))
|
||||
}
|
||||
|
||||
// Execute function
|
||||
result, err := backend.Execute(ctx, function, execution.Input)
|
||||
s.logger.Debug("Starting function execution in runtime backend (async)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()),
|
||||
zap.String("runtime", string(function.Runtime)))
|
||||
result, err := backend.Execute(execCtx, function, execution.Input)
|
||||
s.logger.Debug("Function execution completed in runtime backend (async)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("function_id", function.ID.String()),
|
||||
zap.Error(err))
|
||||
if err != nil {
|
||||
s.logger.Error("Async function execution failed",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.Error(err))
|
||||
execution.Status = domain.StatusFailed
|
||||
execution.Error = fmt.Sprintf("execution failed: %v", err)
|
||||
// Check if this was a timeout error
|
||||
if execCtx.Err() == context.DeadlineExceeded {
|
||||
s.logger.Debug("Async execution timed out (context deadline exceeded)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.Duration("timeout", function.Timeout.Duration),
|
||||
zap.Error(execCtx.Err()))
|
||||
execution.Status = domain.StatusTimeout
|
||||
execution.Error = fmt.Sprintf("function execution timed out after %v", function.Timeout.Duration)
|
||||
} else {
|
||||
s.logger.Debug("Async execution failed (non-timeout error)",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.Error(err))
|
||||
execution.Status = domain.StatusFailed
|
||||
execution.Error = fmt.Sprintf("execution failed: %v", err)
|
||||
}
|
||||
s.updateExecutionComplete(ctx, execution)
|
||||
return
|
||||
}
|
||||
@ -193,8 +291,25 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
|
||||
execution.Duration = result.Duration
|
||||
execution.MemoryUsed = result.MemoryUsed
|
||||
|
||||
// Check if the result indicates a timeout
|
||||
if result.Error != "" {
|
||||
execution.Status = domain.StatusFailed
|
||||
s.logger.Debug("Execution result contains error",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("error", result.Error))
|
||||
if strings.Contains(result.Error, "timed out") {
|
||||
s.logger.Debug("Result indicates timeout - setting status to timeout",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("error", result.Error))
|
||||
execution.Status = domain.StatusTimeout
|
||||
} else {
|
||||
s.logger.Debug("Result indicates failure - setting status to failed",
|
||||
zap.String("execution_id", execution.ID.String()),
|
||||
zap.String("error", result.Error))
|
||||
execution.Status = domain.StatusFailed
|
||||
}
|
||||
} else {
|
||||
s.logger.Debug("Execution successful - no errors",
|
||||
zap.String("execution_id", execution.ID.String()))
|
||||
}
|
||||
|
||||
s.updateExecutionComplete(ctx, execution)
|
||||
|
||||
Reference in New Issue
Block a user