Repository: skypilot-org/skypilot
Pull request: k8s logtail
<!-- Describe the changes in this PR -->

Closes #3429

This swaps the container sleep with a logtailing process so that pod logs now show task logs. These are retrievable via `kubectl logs <POD_NAME>` and also written to `/var/log/pods/*/*/*.log` on the node disk, so it should integrate with most k8s logging solutions out of the box (OpenTelemetry, Fluentd, Filebeats, etc.)

Implemented as a watcher process that does `tail -f` if there's an open writing file handle on the task log by checking `lsof` and closes the `tail -f` program if there are no such writers.

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [x] All smoke tests: `pytest tests/test_smoke.py` 
- [x] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_container_logs_multinode_kubernetes` 
- [x] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_container_logs_two_jobs_kubernetes` 
- [x] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_container_logs_two_simultaneous_jobs_kubernetes` 


Problem statement:
[k8s] Log SkyPilot output to container stdout
SkyPilot pods are run with `sleep infinity` to let SkyPilot runtime manage the lifecycle of the pod. As a result, container logs are empty `kubectl get logs`.

This causes problems when integrating with other log management systems running on the cluster, which directly scrape from the container logs. 

From user(s):
> We are using FileBeat for logs collection. Basically, all the pods' logs are automatically wrote into the dir /var/log/containers/ by k8s (the dir is on the host where the pod is running on), and we just configure this dir into FileBeat.
I found that when we use Sky, all the log files of LLM instances under /var/log/containers/ are just empty because Sky redirects logs
> Another user stated they use external logging system to collect logs of pods

Can we replace the `sleep infinity` command with something which monitors `~/sky_logs` and runs `tail` on any files that get created. Will need to handle multiple tasks executing simultaneously since their logs may get interleaved. 

## Workaround for now
Mount the `~/sky_logs`  to a local hostPath on the host machine. For example, if you add this to your `~/.sky/config.yaml`:
```
kubernetes:
  pod_config:
    spec:
      containers:
        - volumeMounts:
            - mountPath: /home/sky/sky_logs
              name: sky-logs-volume
              readOnly: false
      volumes:
        - name: sky-logs-volume
          hostPath:
            path: /var/log/sky_logs
            type: Directory
```

it will write all logs to /var/log/sky_logs on the host machine. Then you can add /var/log/sky_logs as a scrape path on your logging system. This is similar to how k8s writes all logs to /var/log/containers/.

For example, here's the directory structure that gets created when running two sky clusters independently:
```
/var/log/sky_logs$ tree # This is on the host machine
.
├── sky-2024-04-04-12-32-13-597156
│   ├── run.log
│   └── tasks
│       └── run.log
└── sky-2024-04-04-12-33-24-369770
    ├── run.log
    └── tasks
        └── run.log
```
Make sure you create the dir and make sure it has appropriate permissions (`mkdir -p /var/log/sky_logs; chmod -R 777 /var/log/sky_logs`) before mounting it.

Hints from issue discussion:
[This commit](https://github.com/skypilot-org/skypilot/commit/b95c66491834272b01d68efc955ad4fd9171e4a8) has an alternative solution which routes the stdout to a shared global log file which is tail'd in the container run command. While this also works, it adds the overhead of doubling log size. 

Perhaps we could have a lightweight daemon which monitors ~/sky_logs for new files and prints it to container stdout? Will have to be careful about knowing when to close a file etc depending on when the task completes.

Commit message: lint

Patch to review:
diff --git a/sky/templates/kubernetes-ray.yml.j2 b/sky/templates/kubernetes-ray.yml.j2
index bd4bafd43d5..8ffb8da8ddb 100644
--- a/sky/templates/kubernetes-ray.yml.j2
+++ b/sky/templates/kubernetes-ray.yml.j2
@@ -316,8 +316,48 @@ available_node_types:
           image: {{image_id}}
           # Do not change this command - it keeps the pod alive until it is
           # explicitly killed.
-          command: ["/bin/bash", "-c", "--"]
-          args: ['trap : TERM INT; sleep infinity & wait;']
+          command: ["/bin/bash", "-c"]
+          args: 
+            - |
+              FILE_PATTERN="~/sky_logs/*/tasks/*.log"
+              while ! ls $(eval echo $FILE_PATTERN) 1> /dev/null 2>&1; do
+                sleep 0.1
+              done
+              # Tails file and checks every 5 sec for
+              # open file handlers with write access
+              # closes if none exist
+              monitor_file() {
+                tail -n 0 -f $file &
+                TAIL_PID=$!
+                while kill -0 $TAIL_PID 2> /dev/null; do
+                  # only two PIDs should be accessing the file
+                  # the log appender and log tailer
+                  if [ $(lsof -w $file | wc -l) -lt 3 ]; then
+                    kill $TAIL_PID
+                    break
+                  fi
+                  sleep 30
+                done
+              }
+
+              # Keep track of already monitored files
+              already_monitored=""
+
+              # Infinite loop to continuously check for new files
+              while true; do
+                for file in $(eval echo $FILE_PATTERN); do
+                  if echo $already_monitored | grep -q $file; then
+                    # File is already being monitored
+                    continue
+                  fi
+                  
+                  # Monitor the new file
+                  monitor_file $file &
+                  already_monitored="${already_monitored} ${file}"
+                done
+                sleep 0.1
+              done
+
           ports:
           - containerPort: 22  # Used for SSH
           - containerPort: {{ray_port}}  # Redis port
@@ -365,7 +405,7 @@ setup_commands:
   # Line 'sudo grep ..': set the number of threads per process to unlimited to avoid ray job submit stucking issue when the number of running ray jobs increase.
   # Line 'mkdir -p ..': disable host key check
   # Line 'python3 -c ..': patch the buggy ray files and enable `-o allow_other` option for `goofys`
-  - sudo DEBIAN_FRONTEND=noninteractive apt install gcc patch pciutils rsync fuse curl -y;
+  - sudo DEBIAN_FRONTEND=noninteractive apt install lsof gcc patch pciutils rsync fuse curl -y;
     mkdir -p ~/.ssh; touch ~/.ssh/config;
     {%- for initial_setup_command in initial_setup_commands %}
     {{ initial_setup_command }}
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 0413447dcec..3dd902ab025 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -2044,6 +2044,91 @@ def test_task_labels_kubernetes():
         run_one_test(test)
 
 
+# ---------- Container logs from task on Kubernetes ----------
+@pytest.mark.kubernetes
+def test_container_logs_multinode_kubernetes():
+    name = _get_cluster_name()
+    task_yaml = 'tests/test_yamls/test_k8s_logs.yaml'
+    head_logs = ('kubectl get pods '
+                 f' | grep {name} |  grep head | '
+                 " awk '{print $1}' | xargs -I {} kubectl logs {}")
+    worker_logs = ('kubectl get pods '
+                   f' | grep {name} |  grep worker |'
+                   " awk '{print $1}' | xargs -I {} kubectl logs {}")
+    with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w') as f:
+        test = Test(
+            'container_logs_multinode_kubernetes',
+            [
+                f'sky launch -y -c {name} {task_yaml} --num-nodes 2',
+                f'{head_logs} | wc -l | grep 9',
+                f'{worker_logs} | wc -l | grep 9',
+            ],
+            f'sky down -y {name}',
+        )
+        run_one_test(test)
+
+
+@pytest.mark.kubernetes
+def test_container_logs_two_jobs_kubernetes():
+    name = _get_cluster_name()
+    task_yaml = 'tests/test_yamls/test_k8s_logs.yaml'
+    pod_logs = ('kubectl get pods '
+                f' | grep {name} |  grep head |'
+                " awk '{print $1}' | xargs -I {} kubectl logs {}")
+    with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w') as f:
+        test = Test(
+            'test_container_logs_two_jobs_kubernetes',
+            [
+                f'sky launch -y -c {name} {task_yaml}',
+                f'{pod_logs} | wc -l | grep 9',
+                f'sky launch -y -c {name} {task_yaml}',
+                f'{pod_logs} | wc -l | grep 18',
+                f'{pod_logs} | grep 1 | wc -l | grep 2',
+                f'{pod_logs} | grep 2 | wc -l | grep 2',
+                f'{pod_logs} | grep 3 | wc -l | grep 2',
+                f'{pod_logs} | grep 4 | wc -l | grep 2',
+                f'{pod_logs} | grep 5 | wc -l | grep 2',
+                f'{pod_logs} | grep 6 | wc -l | grep 2',
+                f'{pod_logs} | grep 7 | wc -l | grep 2',
+                f'{pod_logs} | grep 8 | wc -l | grep 2',
+                f'{pod_logs} | grep 9 | wc -l | grep 2',
+            ],
+            f'sky down -y {name}',
+        )
+        run_one_test(test)
+
+
+@pytest.mark.kubernetes
+def test_container_logs_two_simultaneous_jobs_kubernetes():
+    name = _get_cluster_name()
+    task_yaml = 'tests/test_yamls/test_k8s_logs.yaml '
+    pod_logs = ('kubectl get pods '
+                f' | grep {name} |  grep head |'
+                " awk '{print $1}' | xargs -I {} kubectl logs {}")
+    with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w') as f:
+        test = Test(
+            'test_container_logs_two_simultaneous_jobs_kubernetes',
+            [
+                f'sky launch -y -c {name}',
+                f'sky exec -c {name} -d {task_yaml}',
+                f'sky exec -c {name} -d {task_yaml}',
+                'sleep 5',
+                f'{pod_logs} | wc -l | grep 18',
+                f'{pod_logs} | grep 1 | wc -l | grep 2',
+                f'{pod_logs} | grep 2 | wc -l | grep 2',
+                f'{pod_logs} | grep 3 | wc -l | grep 2',
+                f'{pod_logs} | grep 4 | wc -l | grep 2',
+                f'{pod_logs} | grep 5 | wc -l | grep 2',
+                f'{pod_logs} | grep 6 | wc -l | grep 2',
+                f'{pod_logs} | grep 7 | wc -l | grep 2',
+                f'{pod_logs} | grep 8 | wc -l | grep 2',
+                f'{pod_logs} | grep 9 | wc -l | grep 2',
+            ],
+            f'sky down -y {name}',
+        )
+        run_one_test(test)
+
+
 # ---------- Task: n=2 nodes with setups. ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
 @pytest.mark.no_ibm  # IBM cloud currently doesn't provide public image with CUDA
@@ -3140,6 +3225,7 @@ def test_kubernetes_custom_image(image_id):
     run_one_test(test)
 
 
+@pytest.mark.azure
 def test_azure_start_stop_two_nodes():
     name = _get_cluster_name()
     test = Test(
diff --git a/tests/test_yamls/test_k8s_logs.yaml b/tests/test_yamls/test_k8s_logs.yaml
new file mode 100644
index 00000000000..7d5ff368377
--- /dev/null
+++ b/tests/test_yamls/test_k8s_logs.yaml
@@ -0,0 +1,8 @@
+name: test-k8s-logs
+
+run: |
+  for i in $(seq 1 9)
+  do
+    echo "$i"
+    sleep 0.1
+  done


Code review comments:
diff_hunk: @@ -316,8 +316,48 @@ available_node_types:
           image: {{image_id}}
           # Do not change this command - it keeps the pod alive until it is
           # explicitly killed.
-          command: ["/bin/bash", "-c", "--"]
-          args: ['trap : TERM INT; sleep infinity & wait;']
+          command: ["/bin/bash", "-c"]
+          args: 
+            - |
+              FILE_PATTERN="~/sky_logs/*/tasks/*.log"
path: sky/templates/kubernetes-ray.yml.j2
text: @user1:
We may also want to retain the `trap : TERM INT;` to handle termination signals from k8s

@author:
sounds good. i added this back.

diff_hunk: @@ -316,8 +316,48 @@ available_node_types:
           image: {{image_id}}
           # Do not change this command - it keeps the pod alive until it is
           # explicitly killed.
-          command: ["/bin/bash", "-c", "--"]
-          args: ['trap : TERM INT; sleep infinity & wait;']
+          command: ["/bin/bash", "-c"]
+          args: 
+            - |
+              FILE_PATTERN="~/sky_logs/*/tasks/*.log"
path: sky/templates/kubernetes-ray.yml.j2
text: @user1:
If anything goes wrong in this script, the pod will terminate. Can we add safeguards against that? For example, can we wrap this script in a `|| sleep infinity` or something similar?

@author:
done. wrapped the logtail in a function with `|| sleep infinity` fallback.