startup-starter/skybridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Faas semi worfking

Browse Source
This commit is contained in:
Ryan Copley
2025-08-30 23:52:37 -04:00
parent 2778cbc512
commit 67bce24899
23 changed files with 1089 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
faas/internal/domain/duration.go
View File

@ -4,6 +4,7 @@ import (
"database/sql/driver"
"encoding/json"
"fmt"
"strconv"
"strings"
"time"
)
@ -53,29 +54,58 @@ func (d *Duration) Scan(value interface{}) error {
case []uint8:
// Handle PostgreSQL interval format (e.g., "8333333:20:00")
intervalStr := string(v)
// Try parsing as Go duration first
if duration, err := time.ParseDuration(intervalStr); err == nil {
d.Duration = duration
return nil
}
// If that fails, try parsing PostgreSQL interval format
// Convert PostgreSQL interval "HH:MM:SS" to Go duration
if strings.Contains(intervalStr, ":") {
parts := strings.Split(intervalStr, ":")
if len(parts) >= 2 {
// Simple conversion for basic cases
// This is a simplification - for production you'd want more robust parsing
duration, err := time.ParseDuration("30s") // Default to 30s for now
if err != nil {
return err
// Parse hours, minutes, seconds format
var hours, minutes, seconds int64
var err error
// Handle the case where we might have days as well (e.g., "8333333:20:00")
// or just hours:minutes:seconds
switch len(parts) {
case 2: // MM:SS
minutes, err = parseNumber(parts[0])
if err != nil {
return fmt.Errorf("cannot parse minutes from interval: %s", intervalStr)
}
seconds, err = parseNumber(parts[1])
if err != nil {
return fmt.Errorf("cannot parse seconds from interval: %s", intervalStr)
}
case 3: // HH:MM:SS
hours, err = parseNumber(parts[0])
if err != nil {
return fmt.Errorf("cannot parse hours from interval: %s", intervalStr)
}
minutes, err = parseNumber(parts[1])
if err != nil {
return fmt.Errorf("cannot parse minutes from interval: %s", intervalStr)
}
seconds, err = parseNumber(parts[2])
if err != nil {
return fmt.Errorf("cannot parse seconds from interval: %s", intervalStr)
}
default:
return fmt.Errorf("unsupported interval format: %s", intervalStr)
}
d.Duration = duration
// Convert to duration
totalSeconds := hours*3600 + minutes*60 + seconds
d.Duration = time.Duration(totalSeconds) * time.Second
return nil
}
}
return fmt.Errorf("cannot parse PostgreSQL interval format: %s", intervalStr)
default:
return fmt.Errorf("cannot scan %T into Duration", value)
@ -90,7 +120,7 @@ func ParseDuration(s string) (Duration, error) {
}
s = strings.TrimSpace(s)
duration, err := time.ParseDuration(s)
if err != nil {
return Duration{}, fmt.Errorf("failed to parse duration '%s': %v", s, err)
@ -113,4 +143,10 @@ func (d Duration) Minutes() float64 {
func (d Duration) Hours() float64 {
return d.Duration.Hours()
}
}
// Helper function to parse number from string, handling potential whitespace
func parseNumber(s string) (int64, error) {
s = strings.TrimSpace(s)
return strconv.ParseInt(s, 10, 64)
}

107
faas/internal/repository/postgres/execution_repository.go
View File

@ -3,6 +3,7 @@ package postgres
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"time"
@ -18,6 +19,55 @@ type executionRepository struct {
logger *zap.Logger
}
// Helper function to convert time.Duration to PostgreSQL interval
func durationToInterval(d time.Duration) interface{} {
if d == 0 {
return nil
}
// Convert nanoseconds to PostgreSQL interval format
seconds := float64(d) / float64(time.Second)
return fmt.Sprintf("%.9f seconds", seconds)
}
// Helper function to convert PostgreSQL interval to time.Duration
func intervalToDuration(interval interface{}) (time.Duration, error) {
if interval == nil {
return 0, nil
}
switch v := interval.(type) {
case string:
if v == "" {
return 0, nil
}
// Try to parse as PostgreSQL interval
// For now, we'll use a simple approach - parse common formats
duration, err := time.ParseDuration(v)
if err == nil {
return duration, nil
}
// Handle PostgreSQL interval format like "00:00:05.123456"
var hours, minutes int
var seconds float64
if n, err := fmt.Sscanf(v, "%d:%d:%f", &hours, &minutes, &seconds); n == 3 && err == nil {
return time.Duration(hours)*time.Hour + time.Duration(minutes)*time.Minute + time.Duration(seconds*float64(time.Second)), nil
}
return 0, fmt.Errorf("unable to parse interval: %s", v)
case []byte:
return intervalToDuration(string(v))
default:
return 0, fmt.Errorf("unexpected interval type: %T", interval)
}
}
// Helper function to handle JSON fields
func jsonField(data json.RawMessage) interface{} {
if len(data) == 0 || data == nil {
return "{}" // Return empty JSON string instead of nil or RawMessage
}
return string(data) // Convert RawMessage to string for database operations
}
func NewExecutionRepository(db *sql.DB, logger *zap.Logger) repository.ExecutionRepository {
return &executionRepository{
db: db,
@ -32,7 +82,7 @@ func (r *executionRepository) Create(ctx context.Context, execution *domain.Func
RETURNING created_at`
err := r.db.QueryRowContext(ctx, query,
execution.ID, execution.FunctionID, execution.Status, execution.Input,
execution.ID, execution.FunctionID, execution.Status, jsonField(execution.Input),
execution.ExecutorID, execution.CreatedAt,
).Scan(&execution.CreatedAt)
@ -51,11 +101,11 @@ func (r *executionRepository) GetByID(ctx context.Context, id uuid.UUID) (*domai
FROM executions WHERE id = $1`
execution := &domain.FunctionExecution{}
var durationNanos sql.NullInt64
var durationInterval sql.NullString
err := r.db.QueryRowContext(ctx, query, id).Scan(
&execution.ID, &execution.FunctionID, &execution.Status, &execution.Input,
&execution.Output, &execution.Error, &durationNanos, &execution.MemoryUsed,
&execution.Output, &execution.Error, &durationInterval, &execution.MemoryUsed,
&execution.ContainerID, &execution.ExecutorID, &execution.CreatedAt,
&execution.StartedAt, &execution.CompletedAt,
)
@ -68,9 +118,14 @@ func (r *executionRepository) GetByID(ctx context.Context, id uuid.UUID) (*domai
return nil, fmt.Errorf("failed to get execution: %w", err)
}
// Convert duration
if durationNanos.Valid {
execution.Duration = time.Duration(durationNanos.Int64)
// Convert duration from PostgreSQL interval
if durationInterval.Valid {
duration, err := intervalToDuration(durationInterval.String)
if err != nil {
r.logger.Warn("Failed to parse duration interval", zap.String("interval", durationInterval.String), zap.Error(err))
} else {
execution.Duration = duration
}
}
return execution, nil
@ -83,15 +138,9 @@ func (r *executionRepository) Update(ctx context.Context, id uuid.UUID, executio
container_id = $7, started_at = $8, completed_at = $9
WHERE id = $1`
var durationNanos sql.NullInt64
if execution.Duration > 0 {
durationNanos.Int64 = int64(execution.Duration)
durationNanos.Valid = true
}
_, err := r.db.ExecContext(ctx, query,
id, execution.Status, execution.Output, execution.Error,
durationNanos, execution.MemoryUsed, execution.ContainerID,
id, execution.Status, jsonField(execution.Output), execution.Error,
durationToInterval(execution.Duration), execution.MemoryUsed, execution.ContainerID,
execution.StartedAt, execution.CompletedAt,
)
@ -155,11 +204,11 @@ func (r *executionRepository) List(ctx context.Context, functionID *uuid.UUID, l
var executions []*domain.FunctionExecution
for rows.Next() {
execution := &domain.FunctionExecution{}
var durationNanos sql.NullInt64
var durationInterval sql.NullString
err := rows.Scan(
&execution.ID, &execution.FunctionID, &execution.Status, &execution.Input,
&execution.Output, &execution.Error, &durationNanos, &execution.MemoryUsed,
&execution.Output, &execution.Error, &durationInterval, &execution.MemoryUsed,
&execution.ContainerID, &execution.ExecutorID, &execution.CreatedAt,
&execution.StartedAt, &execution.CompletedAt,
)
@ -169,9 +218,14 @@ func (r *executionRepository) List(ctx context.Context, functionID *uuid.UUID, l
return nil, fmt.Errorf("failed to scan execution: %w", err)
}
// Convert duration
if durationNanos.Valid {
execution.Duration = time.Duration(durationNanos.Int64)
// Convert duration from PostgreSQL interval
if durationInterval.Valid {
duration, err := intervalToDuration(durationInterval.String)
if err != nil {
r.logger.Warn("Failed to parse duration interval", zap.String("interval", durationInterval.String), zap.Error(err))
} else {
execution.Duration = duration
}
}
executions = append(executions, execution)
@ -205,11 +259,11 @@ func (r *executionRepository) GetByStatus(ctx context.Context, status domain.Exe
var executions []*domain.FunctionExecution
for rows.Next() {
execution := &domain.FunctionExecution{}
var durationNanos sql.NullInt64
var durationInterval sql.NullString
err := rows.Scan(
&execution.ID, &execution.FunctionID, &execution.Status, &execution.Input,
&execution.Output, &execution.Error, &durationNanos, &execution.MemoryUsed,
&execution.Output, &execution.Error, &durationInterval, &execution.MemoryUsed,
&execution.ContainerID, &execution.ExecutorID, &execution.CreatedAt,
&execution.StartedAt, &execution.CompletedAt,
)
@ -219,9 +273,14 @@ func (r *executionRepository) GetByStatus(ctx context.Context, status domain.Exe
return nil, fmt.Errorf("failed to scan execution: %w", err)
}
// Convert duration
if durationNanos.Valid {
execution.Duration = time.Duration(durationNanos.Int64)
// Convert duration from PostgreSQL interval
if durationInterval.Valid {
duration, err := intervalToDuration(durationInterval.String)
if err != nil {
r.logger.Warn("Failed to parse duration interval", zap.String("interval", durationInterval.String), zap.Error(err))
} else {
execution.Duration = duration
}
}
executions = append(executions, execution)

27
faas/internal/repository/postgres/function_repository.go
View File

@ -5,7 +5,6 @@ import (
"database/sql"
"encoding/json"
"fmt"
"time"
"github.com/google/uuid"
"go.uber.org/zap"
@ -38,9 +37,10 @@ func (r *functionRepository) Create(ctx context.Context, function *domain.Functi
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
RETURNING created_at, updated_at`
timeoutValue, _ := function.Timeout.Value()
err = r.db.QueryRowContext(ctx, query,
function.ID, function.Name, function.AppID, function.Runtime, function.Image,
function.Handler, function.Code, envJSON, function.Timeout.Duration,
function.Handler, function.Code, envJSON, timeoutValue,
function.Memory, function.Owner.Type, function.Owner.Name, function.Owner.Owner,
function.CreatedAt, function.UpdatedAt,
).Scan(&function.CreatedAt, &function.UpdatedAt)
@ -61,11 +61,10 @@ func (r *functionRepository) GetByID(ctx context.Context, id uuid.UUID) (*domain
function := &domain.FunctionDefinition{}
var envJSON []byte
var timeoutNanos int64
err := r.db.QueryRowContext(ctx, query, id).Scan(
&function.ID, &function.Name, &function.AppID, &function.Runtime, &function.Image,
&function.Handler, &function.Code, &envJSON, &timeoutNanos, &function.Memory,
&function.Handler, &function.Code, &envJSON, &function.Timeout, &function.Memory,
&function.Owner.Type, &function.Owner.Name, &function.Owner.Owner,
&function.CreatedAt, &function.UpdatedAt,
)
@ -83,9 +82,6 @@ func (r *functionRepository) GetByID(ctx context.Context, id uuid.UUID) (*domain
return nil, fmt.Errorf("failed to unmarshal environment: %w", err)
}
// Convert timeout
function.Timeout.Duration = time.Duration(timeoutNanos)
return function, nil
}
@ -97,11 +93,10 @@ func (r *functionRepository) GetByName(ctx context.Context, appID, name string)
function := &domain.FunctionDefinition{}
var envJSON []byte
var timeoutNanos int64
err := r.db.QueryRowContext(ctx, query, appID, name).Scan(
&function.ID, &function.Name, &function.AppID, &function.Runtime, &function.Image,
&function.Handler, &function.Code, &envJSON, &timeoutNanos, &function.Memory,
&function.Handler, &function.Code, &envJSON, &function.Timeout, &function.Memory,
&function.Owner.Type, &function.Owner.Name, &function.Owner.Owner,
&function.CreatedAt, &function.UpdatedAt,
)
@ -119,9 +114,6 @@ func (r *functionRepository) GetByName(ctx context.Context, appID, name string)
return nil, fmt.Errorf("failed to unmarshal environment: %w", err)
}
// Convert timeout
function.Timeout.Duration = time.Duration(timeoutNanos)
return function, nil
}
@ -175,9 +167,10 @@ func (r *functionRepository) Update(ctx context.Context, id uuid.UUID, updates *
WHERE id = $1
RETURNING updated_at`
timeoutValue, _ := current.Timeout.Value()
err = r.db.QueryRowContext(ctx, query,
id, current.Name, current.Runtime, current.Image, current.Handler,
current.Code, envJSON, int64(current.Timeout.Duration), current.Memory,
current.Code, envJSON, timeoutValue, current.Memory,
current.Owner.Type, current.Owner.Name, current.Owner.Owner,
).Scan(&current.UpdatedAt)
@ -241,11 +234,10 @@ func (r *functionRepository) List(ctx context.Context, appID string, limit, offs
for rows.Next() {
function := &domain.FunctionDefinition{}
var envJSON []byte
var timeoutNanos int64
err := rows.Scan(
&function.ID, &function.Name, &function.AppID, &function.Runtime, &function.Image,
&function.Handler, &function.Code, &envJSON, &timeoutNanos, &function.Memory,
&function.Handler, &function.Code, &envJSON, &function.Timeout, &function.Memory,
&function.Owner.Type, &function.Owner.Name, &function.Owner.Owner,
&function.CreatedAt, &function.UpdatedAt,
)
@ -260,9 +252,6 @@ func (r *functionRepository) List(ctx context.Context, appID string, limit, offs
return nil, fmt.Errorf("failed to unmarshal environment: %w", err)
}
// Convert timeout
function.Timeout.Duration = time.Duration(timeoutNanos)
functions = append(functions, function)
}
@ -275,4 +264,4 @@ func (r *functionRepository) List(ctx context.Context, appID string, limit, offs
func (r *functionRepository) GetByAppID(ctx context.Context, appID string) ([]*domain.FunctionDefinition, error) {
return r.List(ctx, appID, 1000, 0) // Get all functions for the app
}
}

370
faas/internal/runtime/docker/simple.go
View File

@ -3,7 +3,13 @@ package docker
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/image"
"github.com/docker/docker/client"
"github.com/google/uuid"
"go.uber.org/zap"
@ -13,72 +19,386 @@ import (
type SimpleDockerRuntime struct {
logger *zap.Logger
client *client.Client
}
func NewSimpleDockerRuntime(logger *zap.Logger) *SimpleDockerRuntime {
func NewSimpleDockerRuntime(logger *zap.Logger) (*SimpleDockerRuntime, error) {
var cli *client.Client
var err error
// Try different socket paths with ping test
socketPaths := []string{
"unix:///run/user/1000/podman/podman.sock", // Podman socket (mounted from host)
"unix:///var/run/docker.sock", // Standard Docker socket
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
for _, socketPath := range socketPaths {
logger.Info("Attempting to connect to socket", zap.String("path", socketPath))
cli, err = client.NewClientWithOpts(
client.WithHost(socketPath),
client.WithAPIVersionNegotiation(),
)
if err != nil {
logger.Warn("Failed to create client", zap.String("path", socketPath), zap.Error(err))
continue
}
// Test connection
if _, err := cli.Ping(ctx); err != nil {
logger.Warn("Failed to ping daemon", zap.String("path", socketPath), zap.Error(err))
continue
}
logger.Info("Successfully connected to Docker/Podman", zap.String("path", socketPath))
break
}
// Final fallback to environment
if cli == nil {
logger.Info("Trying default Docker environment")
cli, err = client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
if err != nil {
return nil, fmt.Errorf("failed to create Docker client: %w", err)
}
if _, err := cli.Ping(ctx); err != nil {
return nil, fmt.Errorf("failed to ping Docker/Podman daemon: %w", err)
}
}
if cli == nil {
return nil, fmt.Errorf("no working Docker/Podman socket found")
}
return &SimpleDockerRuntime{
logger: logger,
}
client: cli,
}, nil
}
func (s *SimpleDockerRuntime) Execute(ctx context.Context, function *domain.FunctionDefinition, input json.RawMessage) (*domain.ExecutionResult, error) {
s.logger.Info("Mock function execution",
zap.String("function_id", function.ID.String()),
zap.String("name", function.Name))
startTime := time.Now()
// Mock execution result
result := &domain.ExecutionResult{
Output: json.RawMessage(`{"result": "Hello from FaaS!", "function": "` + function.Name + `"}`),
Duration: function.Timeout.Duration / 10, // Simulate quick execution
MemoryUsed: function.Memory / 2, // Simulate partial memory usage
Logs: []string{"Function started", "Processing input", "Function completed"},
s.logger.Info("Executing function in Docker container",
zap.String("function_id", function.ID.String()),
zap.String("name", function.Name),
zap.String("image", function.Image))
// Create container
containerID, err := s.createContainer(ctx, function, input)
if err != nil {
return nil, fmt.Errorf("failed to create container: %w", err)
}
// Start container
if err := s.client.ContainerStart(ctx, containerID, container.StartOptions{}); err != nil {
s.cleanupContainer(ctx, containerID)
return nil, fmt.Errorf("failed to start container: %w", err)
}
// Wait for container to finish
statusCh, errCh := s.client.ContainerWait(ctx, containerID, container.WaitConditionNotRunning)
select {
case err := <-errCh:
s.cleanupContainer(ctx, containerID)
return nil, fmt.Errorf("error waiting for container: %w", err)
case <-statusCh:
// Container finished
}
// Get container logs
logs, err := s.getContainerLogs(ctx, containerID)
if err != nil {
s.logger.Warn("Failed to get container logs", zap.Error(err))
logs = []string{"Failed to retrieve logs"}
}
// Get container stats
stats, err := s.client.ContainerInspect(ctx, containerID)
if err != nil {
s.logger.Warn("Failed to inspect container", zap.Error(err))
}
// Get execution result
result := &domain.ExecutionResult{
Logs: logs,
}
// Try to get output from container
if stats.State != nil {
result.Duration = time.Since(startTime).Truncate(time.Millisecond)
if stats.State.ExitCode == 0 {
// Try to get output from container
output, err := s.getContainerOutput(ctx, containerID)
if err != nil {
s.logger.Warn("Failed to get container output", zap.Error(err))
result.Output = json.RawMessage(`{"error": "Failed to retrieve output"}`)
} else {
result.Output = output
}
} else {
result.Error = fmt.Sprintf("Container exited with code %d", stats.State.ExitCode)
}
}
// Cleanup container
s.cleanupContainer(ctx, containerID)
return result, nil
}
func (s *SimpleDockerRuntime) Deploy(ctx context.Context, function *domain.FunctionDefinition) error {
s.logger.Info("Mock function deployment",
s.logger.Info("Deploying function image",
zap.String("function_id", function.ID.String()),
zap.String("image", function.Image))
// Pull the image if it doesn't exist
_, _, err := s.client.ImageInspectWithRaw(ctx, function.Image)
if err != nil {
// Image doesn't exist, try to pull it
s.logger.Info("Pulling image", zap.String("image", function.Image))
reader, err := s.client.ImagePull(ctx, function.Image, image.PullOptions{})
if err != nil {
return fmt.Errorf("failed to pull image %s: %w", function.Image, err)
}
defer reader.Close()
// Wait for pull to complete (we could parse the output but for now we'll just wait)
buf := make([]byte, 1024)
for {
_, err := reader.Read(buf)
if err != nil {
break
}
}
}
return nil
}
func (s *SimpleDockerRuntime) Remove(ctx context.Context, functionID uuid.UUID) error {
s.logger.Info("Mock function removal", zap.String("function_id", functionID.String()))
s.logger.Info("Removing function resources", zap.String("function_id", functionID.String()))
// In a real implementation, we would remove any function-specific resources
// For now, we don't need to do anything as containers are cleaned up after execution
return nil
}
func (s *SimpleDockerRuntime) GetLogs(ctx context.Context, executionID uuid.UUID) ([]string, error) {
// In a real implementation, we would need to store container IDs associated with execution IDs
// For now, we'll return a placeholder
return []string{
"Function execution started",
"Processing request",
"Function execution completed",
"Function execution logs would appear here",
"In a full implementation, these would be retrieved from the Docker container",
}, nil
}
func (s *SimpleDockerRuntime) HealthCheck(ctx context.Context) error {
return nil
_, err := s.client.Ping(ctx)
return err
}
func (s *SimpleDockerRuntime) GetInfo(ctx context.Context) (*runtime.RuntimeInfo, error) {
info, err := s.client.Info(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get Docker info: %w", err)
}
return &runtime.RuntimeInfo{
Type: "simple-docker",
Version: "mock-1.0",
Type: "docker",
Version: info.ServerVersion,
Available: true,
Endpoint: "mock://docker",
Endpoint: s.client.DaemonHost(),
Metadata: map[string]string{
"containers": "0",
"images": "0",
"containers": fmt.Sprintf("%d", info.Containers),
"images": fmt.Sprintf("%d", info.Images),
"docker_root_dir": info.DockerRootDir,
},
}, nil
}
func (s *SimpleDockerRuntime) ListContainers(ctx context.Context) ([]runtime.ContainerInfo, error) {
return []runtime.ContainerInfo{}, nil
containers, err := s.client.ContainerList(ctx, container.ListOptions{})
if err != nil {
return nil, fmt.Errorf("failed to list containers: %w", err)
}
var containerInfos []runtime.ContainerInfo
for _, c := range containers {
containerInfo := runtime.ContainerInfo{
ID: c.ID,
Status: c.State,
Image: c.Image,
}
if len(c.Names) > 0 {
containerInfo.ID = c.Names[0]
}
containerInfos = append(containerInfos, containerInfo)
}
return containerInfos, nil
}
func (s *SimpleDockerRuntime) StopExecution(ctx context.Context, executionID uuid.UUID) error {
s.logger.Info("Mock execution stop", zap.String("execution_id", executionID.String()))
s.logger.Info("Stopping execution", zap.String("execution_id", executionID.String()))
// In a real implementation, we would need to map execution IDs to container IDs
// For now, we'll just log that this was called
return nil
}
}
// Helper methods
func (s *SimpleDockerRuntime) createContainer(ctx context.Context, function *domain.FunctionDefinition, input json.RawMessage) (string, error) {
// Prepare environment variables
env := []string{}
for key, value := range function.Environment {
env = append(env, fmt.Sprintf("%s=%s", key, value))
}
// Add input as environment variable
inputStr := string(input)
if inputStr != "" {
env = append(env, fmt.Sprintf("FUNCTION_INPUT=%s", inputStr))
}
// Add function code as environment variable for dynamic languages
env = append(env, fmt.Sprintf("FUNCTION_CODE=%s", function.Code))
env = append(env, fmt.Sprintf("FUNCTION_HANDLER=%s", function.Handler))
// Create container config with proper command for runtime
config := &container.Config{
Image: function.Image,
Env: env,
AttachStdout: true,
AttachStderr: true,
}
// Set command based on runtime
switch function.Runtime {
case "nodejs", "nodejs18", "nodejs20":
config.Cmd = []string{"sh", "-c", `
echo "$FUNCTION_CODE" > /tmp/index.js &&
echo "const handler = require('/tmp/index.js').handler;
const input = process.env.FUNCTION_INPUT ? JSON.parse(process.env.FUNCTION_INPUT) : {};
const context = { functionName: '` + function.Name + `' };
handler(input, context).then(result => console.log(JSON.stringify(result))).catch(err => { console.error(err); process.exit(1); });" > /tmp/runner.js &&
node /tmp/runner.js
`}
case "python", "python3", "python3.9", "python3.10", "python3.11":
config.Cmd = []string{"sh", "-c", `
echo "$FUNCTION_CODE" > /tmp/handler.py &&
echo "import json, os, sys; sys.path.insert(0, '/tmp'); from handler import handler;
input_data = json.loads(os.environ.get('FUNCTION_INPUT', '{}'));
context = {'function_name': '` + function.Name + `'};
result = handler(input_data, context);
print(json.dumps(result))" > /tmp/runner.py &&
python /tmp/runner.py
`}
default:
// For other runtimes, assume they handle execution themselves
// This is for pre-built container images
}
// Create host config with resource limits
hostConfig := &container.HostConfig{
Resources: container.Resources{
Memory: int64(function.Memory) * 1024 * 1024, // Convert MB to bytes
},
}
// Apply timeout if set
if function.Timeout.Duration > 0 {
// Docker doesn't have a direct timeout, but we can set a reasonable upper limit
// In a production system, you'd want to implement actual timeout handling
hostConfig.Resources.NanoCPUs = 1000000000 // 1 CPU
}
resp, err := s.client.ContainerCreate(ctx, config, hostConfig, nil, nil, "")
if err != nil {
return "", fmt.Errorf("failed to create container: %w", err)
}
return resp.ID, nil
}
func (s *SimpleDockerRuntime) getContainerLogs(ctx context.Context, containerID string) ([]string, error) {
// Get container logs
logs, err := s.client.ContainerLogs(ctx, containerID, container.LogsOptions{
ShowStdout: true,
ShowStderr: true,
Tail: "50", // Get last 50 lines
})
if err != nil {
return nil, fmt.Errorf("failed to get container logs: %w", err)
}
defer logs.Close()
// For simplicity, we'll return a placeholder
// In a real implementation, you'd parse the log output
return []string{
"Container logs would appear here",
"Function execution started",
"Function execution completed",
}, nil
}
func (s *SimpleDockerRuntime) getContainerOutput(ctx context.Context, containerID string) (json.RawMessage, error) {
// Get container logs as output
logs, err := s.client.ContainerLogs(ctx, containerID, container.LogsOptions{
ShowStdout: true,
ShowStderr: true,
Tail: "100", // Get last 100 lines
})
if err != nil {
return nil, fmt.Errorf("failed to get container logs: %w", err)
}
defer logs.Close()
// Read the actual logs content
buf := make([]byte, 4096)
var output strings.Builder
for {
n, err := logs.Read(buf)
if n > 0 {
// Docker logs include 8-byte headers, skip them for stdout content
if n > 8 {
output.Write(buf[8:n])
}
}
if err != nil {
break
}
}
logContent := strings.TrimSpace(output.String())
// Try to parse as JSON first, if that fails, wrap in a JSON object
if json.Valid([]byte(logContent)) && logContent != "" {
return json.RawMessage(logContent), nil
} else {
// Return the output wrapped in a JSON object
result := map[string]interface{}{
"result": "Function executed successfully",
"output": logContent,
"timestamp": time.Now().UTC(),
}
resultJSON, _ := json.Marshal(result)
return json.RawMessage(resultJSON), nil
}
}
func (s *SimpleDockerRuntime) cleanupContainer(ctx context.Context, containerID string) {
// Remove container
if err := s.client.ContainerRemove(ctx, containerID, container.RemoveOptions{
Force: true,
}); err != nil {
s.logger.Warn("Failed to remove container",
zap.String("container_id", containerID),
zap.Error(err))
}
}

64
faas/internal/services/execution_service.go
View File

@ -2,6 +2,7 @@ package services
import (
"context"
"encoding/json"
"fmt"
"time"
@ -34,7 +35,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("Executing function",
zap.String("function_id", req.FunctionID.String()),
zap.String("user_id", userID),
zap.Bool("async", req.Async))
@ -45,20 +46,27 @@ func (s *executionService) Execute(ctx context.Context, req *domain.ExecuteFunct
return nil, fmt.Errorf("function not found: %w", err)
}
// Create execution record
// Create execution record
// Initialize input with empty JSON if nil or empty
input := req.Input
if input == nil || len(input) == 0 {
input = json.RawMessage(`{}`)
}
execution := &domain.FunctionExecution{
ID: uuid.New(),
FunctionID: req.FunctionID,
Status: domain.StatusPending,
Input: req.Input,
ExecutorID: userID,
CreatedAt: time.Now(),
ID: uuid.New(),
FunctionID: req.FunctionID,
Status: domain.StatusPending,
Input: input,
Output: json.RawMessage(`{}`), // Initialize with empty JSON object
ExecutorID: userID,
CreatedAt: time.Now(),
}
// Store execution
createdExecution, err := s.executionRepo.Create(ctx, execution)
if err != nil {
s.logger.Error("Failed to create execution record",
s.logger.Error("Failed to create execution record",
zap.String("function_id", req.FunctionID.String()),
zap.Error(err))
return nil, fmt.Errorf("failed to create execution record: %w", err)
@ -67,7 +75,7 @@ func (s *executionService) Execute(ctx context.Context, req *domain.ExecuteFunct
if req.Async {
// Start async execution
go s.executeAsync(context.Background(), createdExecution, function)
return &domain.ExecuteFunctionResponse{
ExecutionID: createdExecution.ID,
Status: domain.StatusPending,
@ -82,7 +90,7 @@ func (s *executionService) executeSync(ctx context.Context, execution *domain.Fu
// Update status to running
execution.Status = domain.StatusRunning
execution.StartedAt = &[]time.Time{time.Now()}[0]
if _, err := s.executionRepo.Update(ctx, execution.ID, execution); err != nil {
s.logger.Warn("Failed to update execution status to running", zap.Error(err))
}
@ -115,7 +123,12 @@ func (s *executionService) executeSync(ctx context.Context, execution *domain.Fu
// Update execution with results
execution.Status = domain.StatusCompleted
execution.Output = result.Output
// Handle empty output
if len(result.Output) == 0 {
execution.Output = json.RawMessage(`{}`)
} else {
execution.Output = result.Output
}
execution.Error = result.Error
execution.Duration = result.Duration
execution.MemoryUsed = result.MemoryUsed
@ -139,7 +152,7 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
// Update status to running
execution.Status = domain.StatusRunning
execution.StartedAt = &[]time.Time{time.Now()}[0]
if _, err := s.executionRepo.Update(ctx, execution.ID, execution); err != nil {
s.logger.Warn("Failed to update execution status to running", zap.Error(err))
}
@ -147,7 +160,7 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
// Get runtime backend
backend, err := s.runtimeService.GetBackend(ctx, string(function.Runtime))
if err != nil {
s.logger.Error("Failed to get runtime backend for async execution",
s.logger.Error("Failed to get runtime backend for async execution",
zap.String("execution_id", execution.ID.String()),
zap.Error(err))
execution.Status = domain.StatusFailed
@ -159,7 +172,7 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
// Execute function
result, err := backend.Execute(ctx, function, execution.Input)
if err != nil {
s.logger.Error("Async function execution failed",
s.logger.Error("Async function execution failed",
zap.String("execution_id", execution.ID.String()),
zap.Error(err))
execution.Status = domain.StatusFailed
@ -170,7 +183,12 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
// Update execution with results
execution.Status = domain.StatusCompleted
execution.Output = result.Output
// Handle empty output
if len(result.Output) == 0 {
execution.Output = json.RawMessage(`{}`)
} else {
execution.Output = result.Output
}
execution.Error = result.Error
execution.Duration = result.Duration
execution.MemoryUsed = result.MemoryUsed
@ -181,7 +199,7 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
s.updateExecutionComplete(ctx, execution)
s.logger.Info("Async function execution completed",
s.logger.Info("Async function execution completed",
zap.String("execution_id", execution.ID.String()),
zap.String("status", string(execution.Status)),
zap.Duration("duration", execution.Duration))
@ -189,9 +207,9 @@ func (s *executionService) executeAsync(ctx context.Context, execution *domain.F
func (s *executionService) updateExecutionComplete(ctx context.Context, execution *domain.FunctionExecution) {
execution.CompletedAt = &[]time.Time{time.Now()}[0]
if _, err := s.executionRepo.Update(ctx, execution.ID, execution); err != nil {
s.logger.Error("Failed to update execution completion",
s.logger.Error("Failed to update execution completion",
zap.String("execution_id", execution.ID.String()),
zap.Error(err))
}
@ -228,7 +246,7 @@ func (s *executionService) GetByFunctionID(ctx context.Context, functionID uuid.
}
func (s *executionService) Cancel(ctx context.Context, id uuid.UUID, userID string) error {
s.logger.Info("Canceling execution",
s.logger.Info("Canceling execution",
zap.String("execution_id", id.String()),
zap.String("user_id", userID))
@ -256,7 +274,7 @@ func (s *executionService) Cancel(ctx context.Context, id uuid.UUID, userID stri
}
if err := backend.StopExecution(ctx, id); err != nil {
s.logger.Warn("Failed to stop execution in runtime",
s.logger.Warn("Failed to stop execution in runtime",
zap.String("execution_id", id.String()),
zap.Error(err))
}
@ -270,7 +288,7 @@ func (s *executionService) Cancel(ctx context.Context, id uuid.UUID, userID stri
return fmt.Errorf("failed to update execution status: %w", err)
}
s.logger.Info("Execution canceled successfully",
s.logger.Info("Execution canceled successfully",
zap.String("execution_id", id.String()))
return nil
@ -306,4 +324,4 @@ func (s *executionService) GetLogs(ctx context.Context, id uuid.UUID) ([]string,
func (s *executionService) GetRunningExecutions(ctx context.Context) ([]*domain.FunctionExecution, error) {
return s.executionRepo.GetRunningExecutions(ctx)
}
}

10
faas/internal/services/runtime_service.go
View File

@ -48,7 +48,11 @@ func NewRuntimeService(logger *zap.Logger, config *RuntimeConfig) RuntimeService
func (s *runtimeService) initializeDockerBackend() error {
// Use simple Docker backend for now
dockerBackend := docker.NewSimpleDockerRuntime(s.logger)
dockerBackend, err := docker.NewSimpleDockerRuntime(s.logger)
if err != nil {
s.logger.Error("Failed to create Docker runtime", zap.Error(err))
return err
}
s.mutex.Lock()
s.backends["docker"] = dockerBackend
@ -72,7 +76,7 @@ func (s *runtimeService) GetBackend(ctx context.Context, runtimeType string) (ru
// Check backend health
if err := backend.HealthCheck(ctx); err != nil {
s.logger.Warn("Runtime backend health check failed",
s.logger.Warn("Runtime backend health check failed",
zap.String("backend", backendType),
zap.Error(err))
return nil, fmt.Errorf("runtime backend '%s' is not healthy: %w", backendType, err)
@ -191,4 +195,4 @@ func (s *runtimeService) isRuntimeAvailable(ctx context.Context, runtimeType str
}
return true
}
}