Real-World Applications of Firecracker 🔥

#!/bin/bash

# Configuration
VM_ID=$1
BUILD_ID=$2
KERNEL="/var/lib/firecracker/kernel/vmlinux-5.10"
ROOTFS="/var/lib/firecracker/rootfs/debian-ci.ext4"

# Create network interface
ip tuntap add dev tap${VM_ID} mode tap
ip link set tap${VM_ID} up
brctl addif br0 tap${VM_ID}

# Start Firecracker
cat <<EOF > /tmp/vm${VM_ID}-config.json
{
    "boot-source": {
        "kernel_image_path": "${KERNEL}",
        "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
    },
    "drives": [
        {
            "drive_id": "rootfs",
            "path_on_host": "${ROOTFS}",
            "is_root_device": true,
            "is_read_only": false
        }
    ],
    "network-interfaces": [
        {
            "iface_id": "eth0",
            "guest_mac": "AA:FC:00:00:${VM_ID}:01",
            "host_dev_name": "tap${VM_ID}"
        }
    ],
    "machine-config": {
        "vcpu_count": 2,
        "mem_size_mib": 2048
    }
}
EOF

# Run the build in isolated environment
firecracker --api-sock /tmp/firecracker-${VM_ID}.socket --config-file /tmp/vm${VM_ID}-config.json

import os
import json
import subprocess
from flask import Flask, request, jsonify

app = Flask(__name__)

class SandboxManager:
    def __init__(self):
        self.base_kernel = "/var/lib/firecracker/kernel/vmlinux-5.10"
        self.base_rootfs = "/var/lib/firecracker/rootfs/dev-environment.ext4"
        self.vm_configs = {}

    def create_sandbox(self, dev_id, project_type):
        # Create a copy of base rootfs for this developer
        dev_rootfs = f"/var/lib/firecracker/rootfs/dev-{dev_id}.ext4"
        subprocess.run(["cp", self.base_rootfs, dev_rootfs])

        # Configure network
        tap_device = f"tap{dev_id}"
        subprocess.run([
            "ip", "tuntap", "add",
            "dev", tap_device,
            "mode", "tap"
        ])
        subprocess.run(["ip", "link", "set", tap_device, "up"])
        subprocess.run(["brctl", "addif", "br0", tap_device])

        # Create VM configuration
        config = {
            "boot-source": {
                "kernel_image_path": self.base_kernel,
                "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
            },
            "drives": [{
                "drive_id": "rootfs",
                "path_on_host": dev_rootfs,
                "is_root_device": True,
                "is_read_only": False
            }],
            "network-interfaces": [{
                "iface_id": "eth0",
                "guest_mac": f"AA:FC:00:00:{dev_id:02x}:01",
                "host_dev_name": tap_device
            }],
            "machine-config": {
                "vcpu_count": 4,
                "mem_size_mib": 4096
            }
        }

        # Store configuration
        self.vm_configs[dev_id] = config
        return config

sandbox_manager = SandboxManager()

@app.route('/sandbox', methods=['POST'])
def create_sandbox():
    data = request.get_json()
    dev_id = data.get('dev_id')
    project_type = data.get('project_type')

    config = sandbox_manager.create_sandbox(dev_id, project_type)
    return jsonify({"status": "success", "config": config})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

#!/bin/bash

# Configuration
LAB_NAME=$1
TOOLS_LIST=$2

# Create a fresh rootfs for security testing
dd if=/dev/zero of=/tmp/security-lab.ext4 bs=1G count=20
mkfs.ext4 /tmp/security-lab.ext4

# Mount and prepare the rootfs
mkdir -p /mnt/security-lab
mount /tmp/security-lab.ext4 /mnt/security-lab

# Install base system and security tools
debootstrap --include=$(echo $TOOLS_LIST | tr ',' ' ') \
    bullseye /mnt/security-lab http://deb.debian.org/debian/

# Configure network isolation
cat <<EOF > /mnt/security-lab/etc/network/interfaces
auto eth0
iface eth0 inet static
    address 172.16.0.2
    netmask 255.255.255.0
    gateway 172.16.0.1
EOF

# Add security hardening
cat <<EOF > /mnt/security-lab/etc/sysctl.d/99-security.conf
net.ipv4.ip_forward = 0
net.ipv4.conf.all.accept_redirects = 0
net.ipv4.conf.default.accept_redirects = 0
net.ipv4.conf.all.secure_redirects = 0
net.ipv4.conf.default.secure_redirects = 0
net.ipv4.conf.all.accept_source_route = 0
net.ipv4.conf.default.accept_source_route = 0
EOF

# Prepare VM configuration
cat <<EOF > /tmp/security-lab-config.json
{
    "boot-source": {
        "kernel_image_path": "/var/lib/firecracker/kernel/vmlinux-5.10",
        "boot_args": "console=ttyS0 reboot=k panic=1 pci=off ip=172.16.0.2::172.16.0.1:255.255.255.0::eth0:off"
    },
    "drives": [
        {
            "drive_id": "rootfs",
            "path_on_host": "/tmp/security-lab.ext4",
            "is_root_device": true,
            "is_read_only": false
        }
    ],
    "network-interfaces": [
        {
            "iface_id": "eth0",
            "guest_mac": "AA:FC:00:00:00:01",
            "host_dev_name": "tap0"
        }
    ],
    "machine-config": {
        "vcpu_count": 2,
        "mem_size_mib": 4096
    }
}
EOF

import os
import json
import uuid
import asyncio
from aiohttp import web

class FunctionRunner:
    def __init__(self):
        self.vm_pool = []
        self.max_vms = 10
        self.base_config = self._load_base_config()

    def _load_base_config(self):
        return {
            "boot-source": {
                "kernel_image_path": "/var/lib/firecracker/kernel/vmlinux-5.10",
                "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
            },
            "drives": [{
                "drive_id": "rootfs",
                "path_on_host": "/var/lib/firecracker/rootfs/lambda-base.ext4",
                "is_root_device": True,
                "is_read_only": True
            }],
            "machine-config": {
                "vcpu_count": 1,
                "mem_size_mib": 128
            }
        }

    async def create_function_vm(self, function_code):
        vm_id = str(uuid.uuid4())

        # Create overlay filesystem for this function
        overlay_path = f"/tmp/function-{vm_id}.ext4"
        os.system(f"dd if=/dev/zero of={overlay_path} bs=1M count=512")
        os.system(f"mkfs.ext4 {overlay_path}")

        # Configure VM
        config = self.base_config.copy()
        config["drives"].append({
            "drive_id": "overlay",
            "path_on_host": overlay_path,
            "is_root_device": False,
            "is_read_only": False
        })

        # Start VM and return handle
        return await self._start_vm(vm_id, config)

    async def _start_vm(self, vm_id, config):
        # Implementation details for starting VM
        pass

    async def execute_function(self, function_code, event):
        vm = await self.create_function_vm(function_code)
        try:
            result = await vm.invoke(event)
            return result
        finally:
            await vm.cleanup()

app = web.Application()
runner = FunctionRunner()

async def handle_function(request):
    data = await request.json()
    result = await runner.execute_function(
        data['function_code'],
        data['event']
    )
    return web.json_response(result)

app.router.add_post('/functions/invoke', handle_function)

if __name__ == '__main__':
    web.run_app(app)

import asyncio
import docker
from typing import Dict, List
import yaml

class TenantEnvironment:
    def __init__(self, tenant_id: str, config: Dict):
        self.tenant_id = tenant_id
        self.config = config
        self.networks: List[str] = []
        self.vms: List[str] = []

    async def setup(self):
        # Create isolated network for tenant
        network_config = await self._create_network()

        # Setup application components
        await self._setup_components()

        # Configure monitoring
        await self._setup_monitoring()

    async def _create_network(self):
        # Implementation for creating isolated network
        pass

    async def _setup_components(self):
        # Setup different application components in isolated VMs
        pass

    async def _setup_monitoring(self):
        # Setup monitoring for this tenant's environment
        pass

class MultiTenantTestManager:
    def __init__(self, config_file: str):
        self.config = self._load_config(config_file)
        self.tenant_environments: Dict[str, TenantEnvironment] = {}

    def _load_config(self, config_file: str) -> Dict:
        with open(config_file) as f:
            return yaml.safe_load(f)

    async def create_tenant_environment(self, tenant_id: str):
        env = TenantEnvironment(tenant_id, self.config)
        await env.setup()
        self.tenant_environments[tenant_id] = env
        return env

    async def run_tests(self, test_suite: str):
        # Run tests across all tenant environments
        pass

    async def cleanup(self):
        # Cleanup all tenant environments
        pass

import asyncio
import networkx as nx
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class EdgeNode:
    node_id: str
    location: tuple
    capacity: Dict
    connections: List[str]
    vm_config: Dict

class EdgeComputeSimulator:
    def __init__(self, topology_file: str):
        self.topology = self._load_topology(topology_file)
        self.nodes: Dict[str, EdgeNode] = {}
        self.network = nx.Graph()

    def _load_topology(self, topology_file: str):
        # Load network topology from file
        pass

    async def create_edge_node(self, node_config: Dict):
        # Create Firecracker VM for edge node
        node_id = node_config['id']

        # Configure networking for this node
        network_config = self._create_network_config(node_config)

        # Create VM configuration
        vm_config = {
            "boot-source": {
                "kernel_image_path": "/var/lib/firecracker/kernel/vmlinux-5.10",
                "boot_args": f"console=ttyS0 reboot=k panic=1 pci=off ip={network_config['ip']}"
            },
            "drives": [{
                "drive_id": "rootfs",
                "path_on_host": f"/var/lib/firecracker/rootfs/edge-{node_id}.ext4",
                "is_root_device": True,
                "is_read_only": False
            }],
            "network-interfaces": [{
                "iface_id": "eth0",
                "guest_mac": f"AA:FC:00:{node_id:02x}:00:01",
                "host_dev_name": f"tap{node_id}"
            }],
            "machine-config": {
                "vcpu_count": node_config.get('vcpu', 1),
                "mem_size_mib": node_config.get('memory', 512)
            }
        }

        # Create and store node
        node = EdgeNode(
            node_id=node_id,
            location=node_config['location'],
            capacity=node_config['capacity'],
            connections=node_config['connections'],
            vm_config=vm_config
        )
        self.nodes[node_id] = node
        return node

    def _create_network_config(self, node_config: Dict):
        # Create network configuration for edge node
        ip = f"172.16.{node_config['subnet']}.{node_config['host']}/24"
        return {
            "ip": ip,
            "gateway": f"172.16.{node_config['subnet']}.1"
        }

    async def simulate_network_conditions(self):
        # Simulate different network conditions between nodes
        for edge1, edge2 in self.network.edges():
            # Apply network latency and bandwidth constraints
            latency = self.network[edge1][edge2].get('latency', '20ms')
            bandwidth = self.network[edge1][edge2].get('bandwidth', '100mbit')

            os.system(f"tc qdisc add dev {edge1}-{edge2} root netem delay {latency}")
            os.system(f"tc qdisc add dev {edge1}-{edge2} root tbf rate {bandwidth} burst 32kbit latency 400ms")

    async def deploy_application(self, app_config: Dict):
        # Deploy application across edge nodes
        for node_id, deployment in app_config['deployments'].items():
            node = self.nodes[node_id]
            await self._deploy_to_node(node, deployment)

    async def _deploy_to_node(self, node: EdgeNode, deployment: Dict):
        # Implementation for deploying to specific node
        pass

# Example usage
async def main():
    simulator = EdgeComputeSimulator('topology.yaml')

    # Create edge nodes
    nodes = [
        {'id': '01', 'location': (40.7128, -74.0060), 'subnet': 1, 'host': 2},
        {'id': '02', 'location': (34.0522, -118.2437), 'subnet': 1, 'host': 3},
        {'id': '03', 'location': (51.5074, -0.1278), 'subnet': 1, 'host': 4}
    ]

    for node_config in nodes:
        await simulator.create_edge_node(node_config)

    # Simulate network conditions
    await simulator.simulate_network_conditions()

    # Deploy application
    app_config = {
        'deployments': {
            '01': {'components': ['web', 'cache']},
            '02': {'components': ['database']},
            '03': {'components': ['analytics']}
        }
    }
    await simulator.deploy_application(app_config)

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
import yaml
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class ServiceConfig:
    name: str
    version: str
    dependencies: List[str]
    resources: Dict
    config: Dict

class MicroservicesTestEnvironment:
    def __init__(self, config_file: str):
        self.config = self._load_config(config_file)
        self.services: Dict[str, ServiceConfig] = {}
        self.vms: Dict[str, Dict] = {}

    def _load_config(self, config_file: str) -> Dict:
        with open(config_file) as f:
            return yaml.safe_load(f)

    async def create_service_vm(self, service: ServiceConfig):
        """Create a Firecracker VM for a specific microservice"""
        vm_id = f"{service.name}-{service.version}"

        # Create rootfs for service
        rootfs_path = f"/var/lib/firecracker/rootfs/{vm_id}.ext4"
        self._prepare_service_rootfs(service, rootfs_path)

        # Configure VM
        config = {
            "boot-source": {
                "kernel_image_path": "/var/lib/firecracker/kernel/vmlinux-5.10",
                "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
            },
            "drives": [{
                "drive_id": "rootfs",
                "path_on_host": rootfs_path,
                "is_root_device": True,
                "is_read_only": False
            }],
            "network-interfaces": [{
                "iface_id": "eth0",
                "guest_mac": f"AA:FC:00:{hash(vm_id) % 256:02x}:00:01",
                "host_dev_name": f"tap-{vm_id}"
            }],
            "machine-config": {
                "vcpu_count": service.resources.get('cpu', 1),
                "mem_size_mib": service.resources.get('memory', 512)
            }
        }

        self.vms[vm_id] = config
        return config

    def _prepare_service_rootfs(self, service: ServiceConfig, rootfs_path: str):
        # Prepare rootfs with service dependencies and configuration
        pass

    async def setup_service_network(self, service: ServiceConfig):
        """Configure network for service and its dependencies"""
        vm_id = f"{service.name}-{service.version}"

        # Create tap device for service
        os.system(f"ip tuntap add dev tap-{vm_id} mode tap")
        os.system(f"ip link set tap-{vm_id} up")
        os.system(f"brctl addif br0 tap-{vm_id}")

        # Setup network policies for service dependencies
        for dependency in service.dependencies:
            self._setup_network_policy(service.name, dependency)

    def _setup_network_policy(self, service: str, dependency: str):
        # Implement network policies between services
        pass

    async def deploy_services(self):
        """Deploy all services in the configuration"""
        for service_name, service_config in self.config['services'].items():
            service = ServiceConfig(
                name=service_name,
                version=service_config['version'],
                dependencies=service_config.get('dependencies', []),
                resources=service_config.get('resources', {}),
                config=service_config.get('config', {})
            )

            # Create and configure VM for service
            await self.create_service_vm(service)
            await self.setup_service_network(service)

    async def run_integration_tests(self):
        """Run integration tests across services"""
        # Implementation for running tests
        pass

# Example usage
async def main():
    env = MicroservicesTestEnvironment('services-config.yaml')
    await env.deploy_services()
    await env.run_integration_tests()

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
import statistics
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class LoadTestConfig:
    target_url: str
    concurrent_users: int
    ramp_up_time: int
    test_duration: int
    scenarios: List[Dict]

class LoadTestEnvironment:
    def __init__(self, config: LoadTestConfig):
        self.config = config
        self.vms: List[str] = []
        self.results: List[Dict] = []

    async def setup_load_generators(self):
        """Create VMs for load generation"""
        vms_needed = self._calculate_required_vms()

        for i in range(vms_needed):
            vm_id = f"loadgen-{i:03d}"

            # Create VM configuration
            config = {
                "boot-source": {
                    "kernel_image_path": "/var/lib/firecracker/kernel/vmlinux-5.10",
                    "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
                },
                "drives": [{
                    "drive_id": "rootfs",
                    "path_on_host": "/var/lib/firecracker/rootfs/loadgen.ext4",
                    "is_root_device": True,
                    "is_read_only": True
                }],
                "network-interfaces": [{
                    "iface_id": "eth0",
                    "guest_mac": f"AA:FC:00:00:{i:02x}:01",
                    "host_dev_name": f"tap{i}"
                }],
                "machine-config": {
                    "vcpu_count": 2,
                    "mem_size_mib": 1024
                }
            }

            # Create network interface
            os.system(f"ip tuntap add dev tap{i} mode tap")
            os.system(f"ip link set tap{i} up")
            os.system(f"brctl addif br0 tap{i}")

            self.vms.append(vm_id)

    def _calculate_required_vms(self) -> int:
        """Calculate number of VMs needed based on load parameters"""
        users_per_vm = 1000  # Adjust based on your needs
        return max(1, self.config.concurrent_users // users_per_vm)

    async def run_load_test(self):
        """Execute load test across all VMs"""
        # Distribute load across VMs
        users_per_vm = self.config.concurrent_users // len(self.vms)

        tasks = []
        for vm_id in self.vms:
            task = asyncio.create_task(
                self._run_load_test_on_vm(vm_id, users_per_vm)
            )
            tasks.append(task)

        results = await asyncio.gather(*tasks)
        self.results.extend(results)

    async def _run_load_test_on_vm(self, vm_id: str, user_count: int):
        """Run load test on specific VM"""
        # Implementation for running load test on VM
        pass

    def analyze_results(self) -> Dict:
        """Analyze load test results"""
        response_times = [r['response_time'] for r in self.results]

        return {
            'avg_response_time': statistics.mean(response_times),
            'p95_response_time': statistics.quantiles(response_times, n=20)[18],
            'p99_response_time': statistics.quantiles(response_times, n=100)[98],
            'total_requests': len(response_times),
            'success_rate': sum(1 for r in self.results if r['success']) / len(self.results)
        }

# Example usage
async def main():
    config = LoadTestConfig(
        target_url="http://target-service:8080",
        concurrent_users=5000,
        ramp_up_time=300,
        test_duration=3600,
        scenarios=[
            {'name': 'browse_catalog', 'weight': 0.4},
            {'name': 'add_to_cart', 'weight': 0.3},
            {'name': 'checkout', 'weight': 0.3}
        ]
    )

    env = LoadTestEnvironment(config)
    await env.setup_load_generators()
    await env.run_load_test()
    results = env.analyze_results()
    print(f"Load Test Results: {results}")

if __name__ == '__main__':
    asyncio.run(main())