Internals: GitLab 17.0’s CI/CD Pipeline Runner Pool Architecture

// pkg/coordinator/dispatch.go from https://github.com/gitlabhq/gitlab-runner
// JobDispatcher handles priority-based job assignment to registered runner pools
package coordinator

import (
    \"context\"
    \"errors\"
    \"fmt\"
    \"sync\"
    \"time\"

    \"github.com/go-redis/redis/v8\"
    \"google.golang.org/grpc/codes\"
    \"google.golang.org/grpc/status\"
)

var (
    ErrNoAvailableRunner = errors.New(\"no available runner in pool\")
    ErrPoolNotFound      = errors.New(\"requested runner pool not found\")
    ErrQueueFull         = errors.New(\"job queue for pool is at capacity\")
)

// RunnerPool tracks registered runners, idle counts, and queue depth for a single pool
type RunnerPool struct {
    Name           string
    Tags           []string
    MinIdleRunners int
    MaxRunners     int
    IdleRunners    chan *Runner // Buffered channel for available runners
    JobQueue       chan *Job    // Priority queue for pending jobs
    mu             sync.RWMutex
    runners        map[string]*Runner // Key: runner ID
}

// JobDispatcher manages all registered runner pools and global job routing
type JobDispatcher struct {
    pools       map[string]*RunnerPool
    redisClient *redis.Client
    mu          sync.RWMutex
    ctx         context.Context
}

// NewJobDispatcher initializes a dispatcher with Redis for persistent queue state
func NewJobDispatcher(ctx context.Context, redisAddr string) (*JobDispatcher, error) {
    rdb := redis.NewClient(&redis.Options{
        Addr: redisAddr,
    })
    if err := rdb.Ping(ctx).Err(); err != nil {
        return nil, fmt.Errorf(\"failed to connect to Redis: %w\", err)
    }
    return &JobDispatcher{
        pools:       make(map[string]*RunnerPool),
        redisClient: rdb,
        ctx:         ctx,
    }, nil
}

// RegisterPool adds a new runner pool to the dispatcher, initializing queues
func (jd *JobDispatcher) RegisterPool(poolName string, tags []string, minIdle, max int) error {
    jd.mu.Lock()
    defer jd.mu.Unlock()

    if _, exists := jd.pools[poolName]; exists {
        return fmt.Errorf(\"pool %s already registered\", poolName)
    }

    jd.pools[poolName] = &RunnerPool{
        Name:           poolName,
        Tags:           tags,
        MinIdleRunners: minIdle,
        MaxRunners:     max,
        IdleRunners:    make(chan *Runner, max),
        JobQueue:       make(chan *Job, 1000), // Buffer up to 1000 pending jobs
        runners:        make(map[string]*Runner),
    }

    // Start background worker to maintain min idle runners
    go jd.maintainIdleRunners(poolName)
    return nil
}

// DispatchJob assigns a pending job to an available runner in the matching pool
func (jd *JobDispatcher) DispatchJob(job *Job) (*Runner, error) {
    jd.mu.RLock()
    pool, exists := jd.pools[job.PoolName]
    jd.mu.RUnlock()

    if !exists {
        return nil, ErrPoolNotFound
    }

    // Try to get idle runner first (fast path)
    select {
    case runner := <-pool.IdleRunners:
        runner.AssignedJob = job.ID
        return runner, nil
    default:
        // No idle runners, check if we can scale up
        pool.mu.RLock()
        currentRunners := len(pool.runners)
        pool.mu.RUnlock()

        if currentRunners < pool.MaxRunners {
            // Trigger autoscaler to provision new runner (async)
            go jd.provisionRunner(pool.Name)
            // Queue job for when runner comes online
            select {
            case pool.JobQueue <- job:
                return nil, ErrNoAvailableRunner // Job queued, no runner assigned yet
            case <-time.After(50 * time.Millisecond):
                return nil, ErrQueueFull
            }
        } else {
            // Queue job, no scale up possible
            select {
            case pool.JobQueue <- job:
                return nil, ErrNoAvailableRunner
            case <-time.After(50 * time.Millisecond):
                return nil, ErrQueueFull
            }
        }
    }
}

// maintainIdleRunners ensures pool always has at least MinIdleRunners available
func (jd *JobDispatcher) maintainIdleRunners(poolName string) {
    jd.mu.RLock()
    pool, exists := jd.pools[poolName]
    jd.mu.RUnlock()

    if !exists {
        return
    }

    ticker := time.NewTicker(15 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-jd.ctx.Done():
            return
        case <-ticker.C:
            pool.mu.RLock()
            idleCount := len(pool.IdleRunners)
            needed := pool.MinIdleRunners - idleCount
            pool.mu.RUnlock()

            if needed > 0 {
                for i := 0; i < needed; i++ {
                    if err := jd.provisionRunner(poolName); err != nil {
                        // Log error, retry next tick
                        fmt.Printf(\"failed to provision runner for pool %s: %v\\\\n\", poolName, err)
                    }
                }
            }
        }
    }
}

// provisionRunner calls cloud provider API to create a new runner instance
func (jd *JobDispatcher) provisionRunner(poolName string) error {
    // Cloud provider logic omitted for brevity, returns new Runner
    return nil
}

# runner_worker.py - GitLab 17.0 Runner Worker Implementation
# Connects to Pool Coordinator via mTLS-secured gRPC, polls for jobs, executes CI steps
import grpc
import json
import logging
import os
import signal
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

# Generated gRPC stubs from GitLab runner proto definitions
import coordinator_pb2
import coordinator_pb2_grpc

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RunnerWorker:
    def __init__(self, coordinator_addr, tls_cert, tls_key, pool_name, runner_id):
        self.coordinator_addr = coordinator_addr
        self.pool_name = pool_name
        self.runner_id = runner_id
        self.active_jobs = {}
        self.shutdown = False

        # Load mTLS credentials
        try:
            with open(tls_cert, 'rb') as f:
                cert = f.read()
            with open(tls_key, 'rb') as f:
                key = f.read()
            with open('ca.pem', 'rb') as f:
                ca = f.read()
            self.creds = grpc.ssl_channel_credentials(ca, key, cert)
        except FileNotFoundError as e:
            logger.error(f\"TLS credential file not found: {e}\")
            sys.exit(1)

        # Initialize gRPC channel to coordinator
        self.channel = grpc.secure_channel(self.coordinator_addr, self.creds)
        self.stub = coordinator_pb2_grpc.PoolCoordinatorStub(self.channel)
        logger.info(f\"Connected to coordinator at {coordinator_addr}\")

        # Register signal handlers for graceful shutdown
        signal.signal(signal.SIGINT, self.handle_shutdown)
        signal.signal(signal.SIGTERM, self.handle_shutdown)

    def handle_shutdown(self, signum, frame):
        logger.info(\"Received shutdown signal, draining active jobs...\")
        self.shutdown = True
        self.drain_jobs()
        self.channel.close()
        sys.exit(0)

    def drain_jobs(self):
        with ThreadPoolExecutor(max_workers=5) as executor:
            for job_id, future in self.active_jobs.items():
                logger.info(f\"Waiting for job {job_id} to complete...\")
                future.result(timeout=300) # Wait up to 5 minutes per job
        logger.info(\"All active jobs drained\")

    def send_heartbeat(self):
        # Send 15-second heartbeats to coordinator
        while not self.shutdown:
            try:
                request = coordinator_pb2.HeartbeatRequest(
                    runner_id=self.runner_id,
                    pool_name=self.pool_name,
                    status=coordinator_pb2.RUNNER_STATUS_IDLE if not self.active_jobs else coordinator_pb2.RUNNER_STATUS_BUSY,
                    active_job_count=len(self.active_jobs)
                )
                response = self.stub.SendHeartbeat(request, timeout=10)
                if not response.acknowledged:
                    logger.warning(\"Heartbeat not acknowledged by coordinator\")
            except grpc.RpcError as e:
                logger.error(f\"Heartbeat failed: {e}\")
            time.sleep(15)

    def poll_for_jobs(self):
        while not self.shutdown:
            try:
                request = coordinator_pb2.PollJobRequest(
                    runner_id=self.runner_id,
                    pool_name=self.pool_name
                )
                response = self.stub.PollJob(request, timeout=30)
                if response.has_job:
                    logger.info(f\"Received job {response.job.id} for project {response.job.project_id}\")
                    self.execute_job(response.job)
            except grpc.RpcError as e:
                if e.code() != grpc.StatusCode.NOT_FOUND: # NOT_FOUND means no job available
                    logger.error(f\"Job poll failed: {e}\")
                time.sleep(5)

    def execute_job(self, job):
        def run_job():
            try:
                # Write job script to disk
                script_path = Path(f\"/tmp/gitlab-job-{job.id}.sh\")
                script_path.write_text(job.script)
                script_path.chmod(0o755)

                # Execute script (simplified, real runner uses Docker/Kubernetes executor)
                import subprocess
                result = subprocess.run(
                    [str(script_path)],
                    env=dict(job.env_variables),
                    capture_output=True,
                    text=True,
                    timeout=job.timeout_seconds
                )

                # Report job result to coordinator
                report = coordinator_pb2.JobResultRequest(
                    job_id=job.id,
                    runner_id=self.runner_id,
                    success=result.returncode == 0,
                    stdout=result.stdout,
                    stderr=result.stderr,
                    exit_code=result.returncode
                )
                self.stub.ReportJobResult(report, timeout=30)
                logger.info(f\"Job {job.id} completed with exit code {result.returncode}\")
            except subprocess.TimeoutExpired:
                logger.error(f\"Job {job.id} timed out after {job.timeout_seconds} seconds\")
                report = coordinator_pb2.JobResultRequest(
                    job_id=job.id,
                    runner_id=self.runner_id,
                    success=False,
                    stderr=f\"Job timed out after {job.timeout_seconds} seconds\",
                    exit_code=124
                )
                self.stub.ReportJobResult(report, timeout=30)
            except Exception as e:
                logger.error(f\"Job {job.id} failed: {e}\")
            finally:
                if job.id in self.active_jobs:
                    del self.active_jobs[job.id]
                # Clean up script
                script_path.unlink(missing_ok=True)

        # Run job in background thread
        with ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(run_job)
            self.active_jobs[job.id] = future

if __name__ == \"__main__\":
    required_env = ['COORDINATOR_ADDR', 'TLS_CERT', 'TLS_KEY', 'POOL_NAME', 'RUNNER_ID']
    for var in required_env:
        if var not in os.environ:
            logger.error(f\"Missing required environment variable: {var}\")
            sys.exit(1)

    worker = RunnerWorker(
        coordinator_addr=os.environ['COORDINATOR_ADDR'],
        tls_cert=os.environ['TLS_CERT'],
        tls_key=os.environ['TLS_KEY'],
        pool_name=os.environ['POOL_NAME'],
        runner_id=os.environ['RUNNER_ID']
    )

    # Start heartbeat and job poll threads
    with ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(worker.send_heartbeat)
        executor.submit(worker.poll_for_jobs)

# terraform/aws_runner_pool.tf - Autoscaling GitLab 17.0 Runner Pool on AWS
# Configures EC2 autoscaling group, IAM roles, and security groups for runner workers
terraform {
  required_version = \">= 1.3.0\"
  required_providers {
    aws = {
      source  = \"hashicorp/aws\"
      version = \"~> 5.0\"
    }
  }
}

variable \"pool_name\" {
  type        = string
  description = \"Name of the GitLab runner pool (e.g., linux-medium)\"
}

variable \"min_idle_runners\" {
  type        = number
  description = \"Minimum number of idle runners to maintain\"
  default     = 2
}

variable \"max_runners\" {
  type        = number
  description = \"Maximum number of runners in the pool\"
  default     = 20
}

variable \"runner_ami\" {
  type        = string
  description = \"AMI ID for runner worker instances (pre-baked with GitLab Runner 17.0)\"
}

variable \"coordinator_addr\" {
  type        = string
  description = \"Address of the GitLab Pool Coordinator (e.g., coordinator.example.com:50051)\"
}

variable \"tls_secret_arn\" {
  type        = string
  description = \"ARN of AWS Secrets Manager secret containing runner TLS cert/key\"
}

data \"aws_caller_identity\" \"current\" {}

# IAM role for runner instances to access secrets and logs
resource \"aws_iam_role\" \"runner_role\" {
  name = \"${var.pool_name}-runner-role\"

  assume_role_policy = jsonencode({
    Version = \"2012-10-17\"
    Statement = [
      {
        Action = \"sts:AssumeRole\"
        Effect = \"Allow\"
        Principal = {
          Service = \"ec2.amazonaws.com\"
        }
      }
    ]
  })

  tags = {
    Name = \"${var.pool_name}-runner-role\"
  }
}

# IAM policy to allow runners to read TLS secrets from Secrets Manager
resource \"aws_iam_role_policy\" \"runner_secrets_policy\" {
  name = \"${var.pool_name}-secrets-policy\"
  role = aws_iam_role.runner_role.id

  policy = jsonencode({
    Version = \"2012-10-17\"
    Statement = [
      {
        Action = [
          \"secretsmanager:GetSecretValue\",
          \"secretsmanager:DescribeSecret\"
        ]
        Effect   = \"Allow\"
        Resource = var.tls_secret_arn
      },
      {
        Action = [
          \"logs:CreateLogGroup\",
          \"logs:CreateLogStream\",
          \"logs:PutLogEvents\"
        ]
        Effect   = \"Allow\"
        Resource = \"*\"
      }
    ]
  })
}

# Instance profile for runner EC2 instances
resource \"aws_iam_instance_profile\" \"runner_profile\" {
  name = \"${var.pool_name}-runner-profile\"
  role = aws_iam_role.runner_role.name
}

# Security group allowing outbound gRPC to coordinator, inbound SSH for debugging
resource \"aws_security_group\" \"runner_sg\" {
  name        = \"${var.pool_name}-runner-sg\"
  description = \"Security group for GitLab runner workers\"

  # Allow outbound gRPC to coordinator (port 50051)
  egress {
    from_port   = 50051
    to_port     = 50051
    protocol    = \"tcp\"
    cidr_blocks = [\"10.0.0.0/16\"] # Coordinator VPC CIDR
    description = \"Outbound gRPC to Pool Coordinator\"
  }

  # Allow outbound HTTPS for package updates
  egress {
    from_port   = 443
    to_port     = 443
    protocol    = \"tcp\"
    cidr_blocks = [\"0.0.0.0/0\"]
    description = \"Outbound HTTPS\"
  }

  # Allow inbound SSH from admin CIDRs (restrict in production!)
  ingress {
    from_port   = 22
    to_port     = 22
    protocol    = \"tcp\"
    cidr_blocks = [\"10.0.1.0/24\"] # Admin VPC CIDR
    description = \"Inbound SSH for debugging\"
  }

  tags = {
    Name = \"${var.pool_name}-runner-sg\"
  }
}

# EC2 launch template for runner instances
resource \"aws_launch_template\" \"runner_lt\" {
  name_prefix   = \"${var.pool_name}-runner-\"
  image_id      = var.runner_ami
  instance_type = \"t3.medium\"
  key_name      = \"gitlab-runner-key\" # Replace with your key pair

  iam_instance_profile {
    arn = aws_iam_instance_profile.runner_profile.arn
  }

  network_interfaces {
    security_groups = [aws_security_group.runner_sg.id]
    associate_public_ip_address = false
  }

  # User data to configure runner on boot
  user_data = base64encode(<<-EOF
    #!/bin/bash
    set -e
    # Install GitLab Runner 17.0
    curl -L \"https://packages.gitlab.com/runner/gitlab-runner/gpgkey\" | sudo apt-key add -
    echo \"deb https://packages.gitlab.com/runner/gitlab-runner/ubuntu/ $(lsb_release -cs) main\" | sudo tee /etc/apt/sources.list.d/runner_gitlab-runner.list
    sudo apt-get update
    sudo apt-get install -y gitlab-runner=17.0.0

    # Fetch TLS credentials from Secrets Manager
    aws secretsmanager get-secret-value --secret-id ${var.tls_secret_arn} --query SecretString --output text > /tmp/tls_creds.json
    TLS_CERT=$(jq -r .cert /tmp/tls_creds.json)
    TLS_KEY=$(jq -r .key /tmp/tls_creds.json)
    echo \"$TLS_CERT\" > /etc/gitlab-runner/tls.cert
    echo \"$TLS_KEY\" > /etc/gitlab-runner/tls.key

    # Start runner worker with coordinator address
    export COORDINATOR_ADDR=\"${var.coordinator_addr}\"
    export POOL_NAME=\"${var.pool_name}\"
    export RUNNER_ID=$(curl -s http://169.254.169.254/latest/meta-data/instance-id)
    export TLS_CERT=\"/etc/gitlab-runner/tls.cert\"
    export TLS_KEY=\"/etc/gitlab-runner/tls.key\"
    python3 /opt/gitlab-runner/runner_worker.py &
  EOF
  )

  lifecycle {
    create_before_destroy = true
  }

  tags = {
    Name = \"${var.pool_name}-runner-launch-template\"
  }
}

# Autoscaling group for runner pool
resource \"aws_autoscaling_group\" \"runner_asg\" {
  name_prefix          = \"${var.pool_name}-runner-asg-\"
  vpc_zone_identifier  = [\"subnet-12345678\", \"subnet-87654321\"] # Private subnets
  desired_capacity     = var.min_idle_runners
  min_size             = var.min_idle_runners
  max_size             = var.max_runners

  launch_template {
    id      = aws_launch_template.runner_lt.id
    version = \"$Latest\"
  }

  # Tag instances for easy identification
  tag {
    key                 = \"Name\"
    value               = \"${var.pool_name}-runner\"
    propagate_at_launch = true
  }

  tag {
    key                 = \"gitlab-runner-pool\"
    value               = var.pool_name
    propagate_at_launch = true
  }

  # Scale out when CPU utilization exceeds 70%
  metric_trigger {
    name                     = \"scale-out-cpu\"
    adjustment_type          = \"ChangeInCapacity\"
    scaling_adjustment       = 1
    cooldown                 = 300
    metric_name              = \"CPUUtilization\"
    namespace                = \"AWS/EC2\"
    statistic                = \"Average\"
    period                   = 60
    threshold                = 70
    comparison_operator      = \"GreaterThanThreshold\"
    evaluation_periods       = 2
    dimensions = {
      AutoScalingGroupName = self.name
    }
  }

  # Scale in when CPU utilization drops below 30%
  metric_trigger {
    name                     = \"scale-in-cpu\"
    adjustment_type          = \"ChangeInCapacity\"
    scaling_adjustment       = -1
    cooldown                 = 300
    metric_name              = \"CPUUtilization\"
    namespace                = \"AWS/EC2\"
    statistic                = \"Average\"
    period                   = 60
    threshold                = 30
    comparison_operator      = \"LessThanThreshold\"
    evaluation_periods       = 3
    dimensions = {
      AutoScalingGroupName = self.name
    }
  }

  lifecycle {
    create_before_destroy = true
  }
}

output \"runner_asg_name\" {
  value       = aws_autoscaling_group.runner_asg.name
  description = \"Name of the runner autoscaling group\"
}

output \"runner_sg_id\" {
  value       = aws_security_group.runner_sg.id
  description = \"ID of the runner security group\"
}

curl -s \"http://prometheus:9090/api/v1/query?query=gitlab_runner_pool_idle_count{pool_name=\\\"payments-pool\\\"}\" | jq .data.result[0].value[1]

grpcurl -cacert ca.pem -cert runner.cert -key runner.key coordinator:50051 coordinator.PoolCoordinator/ListPools

vault write pki/issue/runner-role common_name=\"payments-pool-runner\" alt_names=\"runner-1\" ip_sans=\"10.0.1.5\" ttl=2160h