AWS Lambda 직접 구현하기 - 5 (프로세스 종속성, 파일 시스템 공유 문제 해결)

1. VM 실행동안 프로세스 활성화 문제

2. 파일시스템 공유 문제 

 

두 개의 문제를 해결해야 한다. 1번은 생각보다 쉬운데 바로 수정해 보자.

 

 

1. VM 실행동안 프로세스 활성화 문제 수정

Firecracker 소켓을 여는 프로세스를 개별 프로세스로 실행하고, 일정 시간이 지나면 해당 프로세스가 종료되게 설정하면 된다. 

 

private void firecrackerStart(Map<String, String> env) throws IOException {
    ProcessBuilder firecrackerStartBuilder = new ProcessBuilder(FIRECRACKER_PATH + "/firecracker",
            "--api-sock",
            env.get("API_SOCKET"));
 
    firecrackerStartBuilder.environment().putAll(env);
    firecrackerStartBuilder.start();
}

 

이거는 process.waitFor() 할 필요가 없다. 이유는 start_instance.sh 에서 소켓이 실행되고 있는지 확인하고 있다. 

while [ ! -e "$API_SOCKET" ]; do
    echo "FC $SB_ID still not ready..."
    sleep 0.01
done

 

finally {
    try {
        firecrackerFinish(env);
    } catch (IOException e) {
        // Handle cleanup failure
        System.err.println("Error during firecracker finish: " + e.getMessage());
    }
}
 
private void firecrackerFinish(Map<String, String> env) throws IOException {
        ProcessBuilder firecrackerFinishBuilder = new ProcessBuilder("bash", FIRECARCKER_FINISH_SCRIPT_PATH);
        firecrackerFinishBuilder.environment().putAll(env);
        firecrackerFinishBuilder.start();
    }

 

#!/bin/bash
 
ssh -o StrictHostKeyChecking=no -i $SSH_KEY_PATH root@${FC_IP} << EOF
    sleep 60
    reboot
EOF
 
rm -rf $COPY_ROOTFS

 

Firecarkcer 소켓 실행 스크립트와 microVM 생성 스크립트가 실행된 후 처리해야 할 것은 Firecracker 소켓을 일정 시간 이후 닫는 스크립트다. 그래서 정상 진행 또는 오류로 진행되든 간에 반드시 실행되는 프로세스로 위와 같이 구현했다. 여기서는 60초로 짧게 뒀지만 람다처럼 2시간으로 설정하던지, 적당한 시간으로 설정하는 게 좋아 보인다. 

 

1. 파일시스템 공유 문제 

이것도 어렵지 않은데, 고민했던 부분은 성능이다. 복사하는데 오랜 시간이 걸리면 microVM 생성이 오래걸리게 돼서 응답이 느려지게 된다. 따라서 dd, cp, rsync와 같은 복사 커맨드들을 비교했는데, cp의 --reflink=auto 옵션을 주는 게 가장 빨랐다. 이 옵션은 Copy-on-Write 기능을 사용하게 되는데, 실제 데이터를 복사하는 게 아닌 원본 파일과 새 파일이 동일한 데이터 블록을 공유하고 복사된 파일이 수정되면 새로운 데이터 블록을 생성하게 된다. 디스크 공간도 절약할 수 있고, microVM가 파일시스템 데이터를 공유하는 상황을 방지할 수 있게 된다. start_instance.sh 부분에 아래와 같이 추가했다. 

 

wait_for_cp() {
    while :; do
        # dd 프로세스의 수를 확인
        cp_count=$(pgrep -x -c "cp" || echo 0)
 
        # 프로세스 수가 2 이상이면 대기
        if [ "$cp_count" -ge 1 ]; then
            echo "cp 프로세스가 $cp_count 개 실행 중입니다. 잠시 기다립니다..."
            sleep 0.5  # 0.5초 대기
        else
            echo "cp 프로세스가 종료되었습니다. 계속 진행합니다."
            break
        fi
    done
}
 
~~ 위는 같음
 
  wait_for_cp
  cp --reflink=auto $BASE_ROOTFS $COPY_ROOTFS
  
  # 루트 파일 시스템 설정
    curl_put "/drives/$SB_ID" <<EOF
{
    "drive_id": "$SB_ID",
    "path_on_host": "$COPY_ROOTFS",
    "is_root_device": true,
    "is_read_only": false
}
EOF
~~ 아래는 같음

 

여기서 wait_for_cp를 추가하여 동시에 여러 cp가 실행되는 것을 막았는데, 이유는 동시에 cp를 실행하면 복사를 위해 자원 경쟁이 발생하여 현저하게 속도가 느려지는 것을 확인했다. 2GB 기준으로 단일 실행은 2초인데, 동시에 실행하면 20초까지 늘어나게 된다. 

 

그래서 전체 코드는 아래와 같다.

 

VMManager

package firecracker_application.firecracker_instance.util;
 
import firecracker_application.firecracker_instance.controller.dto.ResourceRequest;
import firecracker_application.firecracker_instance.controller.dto.ResourceResponse;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Value;
 
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
 
@Component
public class VMManager {
 
    @Value("${firecracker.file.ssh-key}")
    private String SSH_KEY_PATH;
 
    @Value("${firecracker.file.rootfs}")
    private String ROOTFS_PATH;
 
    @Value("${firecracker.file.kernel}")
    private String KERNEL_PATH;
 
    @Value("${firecracker.file.logfile}")
    private String LOGFILE_PATH;
 
    @Value("${firecracker.file.start_script}")
    private String START_SCRIPT_PATH;
 
    @Value("${firecracker.file.firecracker_finish_script}")
    private String FIRECARCKER_FINISH_SCRIPT_PATH;
 
    @Value("${firecracker.path}")
    private String FIRECRACKER_PATH;
 
 
    public ResourceResponse instanceStart(final ResourceRequest resourceRequest) {
        List<String> outputLines = new ArrayList<>();
        Map<String, String> env = getEnv(resourceRequest);
 
        try {
            firecrackerStart(env);
            outputLines = executeStartScript(env);
        } catch (Exception e) {
            return new ResourceResponse(Collections.singletonList("Error: " + e.getMessage()));
        } finally {
            try {
                firecrackerFinish(env);
            } catch (IOException e) {
                // Handle cleanup failure
                System.err.println("Error during firecracker finish: " + e.getMessage());
            }
        }
 
        return new ResourceResponse(outputLines);
    }
 
    private List<String> executeStartScript(Map<String, String> env) throws IOException, InterruptedException {
        List<String> outputLines = new ArrayList<>();
        ProcessBuilder instanceStartProcessBuilder = new ProcessBuilder("bash", START_SCRIPT_PATH);
        instanceStartProcessBuilder.environment().putAll(env);
        instanceStartProcessBuilder.redirectErrorStream(true);
 
        Process process = instanceStartProcessBuilder.start();
 
        // Asynchronously read the output
        Thread outputThread = new Thread(() -> readOutput(process, outputLines));
        outputThread.start();
 
        int exitCode = process.waitFor(); // Wait for the process to complete
        outputThread.join(); // Wait for the output thread to finish
 
        if (exitCode != 0) {
            throw new RuntimeException("Process exited with non-zero status: " + exitCode);
        }
 
        return outputLines;
    }
 
    private void readOutput(Process process, List<String> outputLines) {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
            String line;
            boolean withinOutput = false;
 
            while ((line = reader.readLine()) != null) {
                if (line.contains("SCRIPT_OUTPUT_START")) {
                    withinOutput = true; // Start of relevant output
                    continue;
                } else if (line.contains("SCRIPT_OUTPUT_END")) {
                    withinOutput = false; // End of relevant output
                    continue;
                }
                System.out.println(line); 
                if (withinOutput && line.contains("[INFO]")) {
                    outputLines.add(line);
                    // System.out.println(line); // Optional: Print the line
                }
            }
        } catch (IOException e) {
            System.err.println("Error reading output: " + e.getMessage());
        }
    }
 
    private Map<String, String> getEnv(final ResourceRequest resourceRequest) {
        Map<String, String> env = new HashMap<>(System.getenv());
        int sbId = generateRandomSbId();
 
        // Set environment variables
        env.put("VCPU", String.valueOf(Math.max(1, Math.round(resourceRequest.getRequestMemory() / 2048 * 2) / 2)));
        env.put("MEMORY", String.valueOf(resourceRequest.getRequestMemory()));
        env.put("SB_ID", String.valueOf(sbId));
        env.put("TAP_DEV", "tap" + sbId);
        env.put("BUCKET_NAME", resourceRequest.getBucketName());
        env.put("FILE_PATH", resourceRequest.getFilePath());
        env.put("MASK_LONG", "255.255.255.252");
        env.put("MASK_SHORT", "/30");
        setIPAddresses(env, sbId);
        setKernelAndRootFSPaths(env, resourceRequest);
        env.put("LOGFILE", LOGFILE_PATH + sbId + "-log");
        env.put("API_SOCKET", "/tmp/firecracker-sb" + sbId + ".socket");
        env.put("SSH_KEY_PATH", SSH_KEY_PATH);
        env.put("ENV", resourceRequest.getEnv());
        env.put("FIRECRACKER_PATH", FIRECRACKER_PATH);
        
        System.out.println(env.get("BASE_ROOTFS"));
        System.out.println(env.get("COPY_ROOTFS"));
        return env;
    }
 
    private void setIPAddresses(Map<String, String> env, int sbId) {
        env.put("FC_IP", String.format("176.12.0.%d", ((4 * sbId + 1) % 256)));
        env.put("TAP_IP", String.format("176.12.0.%d", ((4 * sbId + 2) % 256)));
        env.put("FC_MAC", String.format("02:FC:00:00:%02X:%02X", sbId / 256, sbId % 256));
    }
 
    private void setKernelAndRootFSPaths(Map<String, String> env, ResourceRequest resourceRequest) {
        env.put("BASE_ROOTFS", ROOTFS_PATH.replace("ubuntu", resourceRequest.getLanguage() + "_ubuntu"));
        env.put("COPY_ROOTFS", ROOTFS_PATH.replace("ubuntu", env.get("SB_ID") + "_" + resourceRequest.getLanguage() + "_ubuntu"));
        env.put("KERNEL", KERNEL_PATH.replace("vmlinux", resourceRequest.getArchitect() + "_vmlinux"));
        env.put("KERNEL_BOOT_ARGS", String.format("console=ttyS0 reboot=k panic=1 pci=off ip=%s::%s:%s::eth0:off",
                env.get("FC_IP"), env.get("TAP_IP"), env.get("MASK_LONG")));
    }
 
    private static int generateRandomSbId() {
        Random random = new Random();
        int id;
        do {
            id = random.nextInt(255);
        } while (isIpAddressInUse(id));
        return id;
    }
 
    private static boolean isIpAddressInUse(int id) {
        String[] command = {"/bin/sh", "-c", "ip addr | grep " + "tap" + id};
 
        try {
            Process process = Runtime.getRuntime().exec(command);
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    if (line.contains("tap" + id)) {
                        return true;  // IP address in use
                    }
                }
            }
 
            int exitCode = process.waitFor();
            if (exitCode != 0) {
                System.out.println("Error checking IP address: " + exitCode);
            }
        } catch (IOException | InterruptedException e) {
            System.out.println("Exception while checking IP address: " + e);
        }
 
        return false;  // IP address not in use
    }
 
    private void firecrackerStart(Map<String, String> env) throws IOException {
        ProcessBuilder firecrackerStartBuilder = new ProcessBuilder(FIRECRACKER_PATH + "/firecracker",
                "--api-sock",
                env.get("API_SOCKET"));
 
        firecrackerStartBuilder.environment().putAll(env);
        firecrackerStartBuilder.start();
    }
 
    private void firecrackerFinish(Map<String, String> env) throws IOException {
        ProcessBuilder firecrackerFinishBuilder = new ProcessBuilder("bash", FIRECARCKER_FINISH_SCRIPT_PATH);
        firecrackerFinishBuilder.environment().putAll(env);
        firecrackerFinishBuilder.start();
    }
    
}

 

start_instance.sh

#!/bin/bash
 
set -euo pipefail
 
# Function to generate a random SB_ID
curl_put() {
    local URL_PATH="$1"
    local OUTPUT RC
    OUTPUT="$("${CURL[@]}" -X PUT --data @- "http://localhost/${URL_PATH#/}" 2>&1)"
    RC="$?"
    if [ "$RC" -ne 0 ]; then
        echo "Error: curl PUT ${URL_PATH} failed with exit code $RC, output:"
        echo "$OUTPUT"
        return 1
    fi
    # Error if output doesn't end with "HTTP 2xx"
    if [[ "$OUTPUT" != *HTTP\ 2[0-9][0-9] ]]; then
        echo "Error: curl PUT ${URL_PATH} failed with non-2xx HTTP status code, output:"
        echo "$OUTPUT"
        return 1
    fi
}
 
wait_for_cp() {
    while :; do
        # dd 프로세스의 수를 확인
        cp_count=$(pgrep -x -c "cp" || echo 0)
 
        # 프로세스 수가 2 이상이면 대기
        if [ "$cp_count" -ge 1 ]; then
            echo "dd 프로세스가 $cp_count 개 실행 중입니다. 잠시 기다립니다..."
            sleep 0.5  # 0.5초 대기
        else
            echo "dd 프로세스가 종료되었습니다. 계속 진행합니다."
            break
        fi
    done
}
 
start_microvm_with_network() {
    while [ ! -e "$API_SOCKET" ]; do
        echo "FC $SB_ID still not ready..."
        sleep 0.01
    done
 
    ip link del "$TAP_DEV" 2> /dev/null || true
    ip tuntap add dev "$TAP_DEV" mode tap
    ip link set dev "$TAP_DEV" up
    ip addr add "${TAP_IP}${MASK_SHORT}" dev "$TAP_DEV"
    sysctl -w net.ipv4.conf.${TAP_DEV}.proxy_arp=1 > /dev/null
    sysctl -w net.ipv6.conf.${TAP_DEV}.disable_ipv6=1 > /dev/null
 
    # 로그 설정
    curl_put "/logger" <<EOF
{
    "level": "Info",
    "log_path": "$LOGFILE",
    "show_level": false,
    "show_log_origin": false
}
EOF
 
    # 머신 설정
    curl_put "/machine-config" <<EOF
{
    "vcpu_count": $VCPU,
    "mem_size_mib": $MEMORY
}
EOF
 
    # 부팅 소스 설정
    curl_put "/boot-source" <<EOF
{
    "kernel_image_path": "$KERNEL",
    "boot_args": "$KERNEL_BOOT_ARGS"
}
EOF
 
    wait_for_cp
    cp --reflink=auto $BASE_ROOTFS $COPY_ROOTFS
 
   # 루트 파일 시스템 설정
    curl_put "/drives/$SB_ID" <<EOF
{
    "drive_id": "$SB_ID",
    "path_on_host": "$COPY_ROOTFS",
    "is_root_device": true,
    "is_read_only": false
}
EOF
 
    # 네트워크 인터페이스 설정
    curl_put "/network-interfaces/$SB_ID" <<EOF
{
    "iface_id": "$SB_ID",
    "guest_mac": "$FC_MAC",
    "host_dev_name": "$TAP_DEV"
}
EOF
 
    # microVM 시작
    curl_put '/actions' <<EOF
{
    "action_type": "InstanceStart"
}
EOF
 
    echo "InstanceStart command executed"
 
    sleep 3
}
 
execute_script_in_microvm() {
    echo "SCRIPT_OUTPUT_START"
    # SSH를 통해 S3에서 스크립트 다운로드 및 실행
    ssh -o StrictHostKeyChecking=no -i $SSH_KEY_PATH root@${FC_IP} << EOF
        echo [INFO] "$FC_IP"
 
        ip route add default via $TAP_IP dev eth0
        echo 'nameserver 8.8.8.8' > /etc/resolv.conf
 
        aws s3 cp s3://$BUCKET_NAME/$FILE_PATH /root
 
        java -Denv="$ENV" $FILE_PATH | while IFS= read -r line; do
            echo "[INFO] \$line"
        done
 
        echo "SCRIPT_OUTPUT_END"
EOF
}
 
CURL=(curl --silent --show-error --header "Content-Type: application/json" --unix-socket "${API_SOCKET}" --write-out "HTTP %{http_code}")
touch $LOGFILE
 
# Start the microVM with networking
start_microvm_with_network
 
execute_script_in_microvm
 
exit 0

 

firecracker_finish.sh

#!/bin/bash
 
ssh -o StrictHostKeyChecking=no -i $SSH_KEY_PATH root@${FC_IP} << EOF
    sleep 60
    reboot
EOF
 
rm -rf $COPY_ROOTFS

 

전체 코드는 github에서 볼 수 있다. 

 

 

개선할 부분

Lambda와 유사하게 만들었지만, 현재는 warm up 부분은 아직 구현하지 못했다. firecracker의 api socker이 실행중일 때 거기로 요청을 보내야 하는데, 이 부분은 RDB 혹은 NOSQL DB를 써야 할 거 같은 생각이 든다. 그래서 해당 부분을 추가할 예정이다.