Updated integration tests and local transfers fix (#831)

* Add local integration tests (`cp_local.py`) 
* Fix transfer completion bug (only consider completions of terminal
operators) which was causing a bug for Azure
* Support local transfers (by falling back to cloud provider APIs)

.github/workflows/integration-test-local.yml

@@ -0,0 +1,95 @@
+name: integration-test-local
+on: [push]
+concurrency: transfer-test
+env:
+  AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
+  AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+  SKYPLANE_USAGE_STATS_ENABLED: 0
+jobs:
+  integration:
+    if: (github.event_name == 'push' && github.ref == 'refs/heads/main') || (github.event_name == 'pull_request')
+    runs-on: ubuntu-latest
+    strategy:
+      max-parallel: 8
+      matrix:
+        pairs:
+          # AWS
+          - aws:us-east-1 local
+          - local aws:us-east-2 
+          # GCP
+          - gcp:us-central1-a local
+          - local gcp:us-east1-a
+          # Azure
+          - azure:westus local
+          - local azure:westus
+    timeout-minutes: 40
+    env:
+      STRATEGY_UUID: itest-${{ github.run_id }}-${{ github.run_attempt }}-${{ strategy.job-index }}
+    steps:
+      - uses: actions/checkout@v1
+      - name: Install poetry
+        run: pipx install poetry
+      - name: Set up Python 3.10
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+          cache: "poetry"
+      - name: Set Poetry config
+        run: |
+          poetry config virtualenvs.in-project false
+          poetry config virtualenvs.path ~/.virtualenvs
+      - name: Install Dependencies
+        run: poetry install -E aws -E azure -E gcp
+        if: steps.cache.outputs.cache-hit != 'true'
+      - id: 'auth'
+        uses: 'google-github-actions/auth@v1'
+        with:
+          credentials_json: '${{ secrets.GCP_CREDENTIALS_JSON }}'
+      - name: Log into Azure
+        uses: azure/login@v1
+        with:
+          creds: '{"clientId":"${{ secrets.AZURE_CLIENT_ID }}","clientSecret":"${{ secrets.AZURE_CLIENT_SECRET }}","subscriptionId":"${{ secrets.AZURE_SUBSCRIPTION_ID }}","tenantId":"${{ secrets.AZURE_TENANT_ID }}"}'
+      - name: Skyplane init
+        run: |
+          poetry run skyplane config set gcp_service_account_name ${{ env.STRATEGY_UUID }}
+          poetry run skyplane config set native_cmd_enabled false
+          cat ~/.skyplane/config
+          poetry run skyplane init -y
+          poetry run skyplane config set usage_stats false
+      - name: Single small file test
+        run: poetry run python tests/integration/cp_local.py ${{ matrix.pairs }} --n-files 1 --file-size-mb 32
+      - name: 128 small files test
+        run: poetry run python tests/integration/cp_local.py ${{ matrix.pairs }} --n-files 128 --file-size-mb 1
+      - name: Single large file test
+        run: poetry run python tests/integration/cp_local.py ${{ matrix.pairs }} --n-files 1 --file-size-mb 2000
+      - name: Cleanup GCP service account
+        if: always()
+        run: gcloud iam service-accounts delete ${{ env.STRATEGY_UUID }}@${{ secrets.GCP_PROJECT_ID }}.iam.gserviceaccount.com
+  deprovision:
+    runs-on: ubuntu-latest
+    if: ${{ always() }}
+    needs: [integration]
+    env:
+      STRATEGY_UUID: itest-d-${{ github.run_id }}-${{ github.run_attempt }}
+    steps:
+      - uses: actions/checkout@v1
+      - name: Set up Python 3.10
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+      - name: Install skyplane from pypi
+        run: pip install skyplane[aws,azure,gcp]
+      - id: 'auth'
+        uses: 'google-github-actions/auth@v1'
+        with:
+          credentials_json: '${{ secrets.GCP_CREDENTIALS_JSON }}'
+      - name: Skyplane init
+        run: |
+          skyplane config set gcp_service_account_name ${{ env.STRATEGY_UUID }}
+          skyplane init -y --disable-config-azure
+          skyplane config set usage_stats false
+      - name: Deprovision
+        run: skyplane deprovision
+      - name: Cleanup GCP service account
+        if: always()
+        run: gcloud iam service-accounts delete --quiet ${{ env.STRATEGY_UUID }}@${{ secrets.GCP_PROJECT_ID }}.iam.gserviceaccount.com

.github/workflows/integration-test-multiple-sizes.yml

@@ -7,28 +7,27 @@ env:
   SKYPLANE_USAGE_STATS_ENABLED: 0
 jobs:
   integration:
-    if: (github.event_name == 'push' && github.ref == 'refs/heads/main') || (github.event_name == 'pull_request')
     runs-on: ubuntu-latest
     strategy:
       max-parallel: 1
       matrix:
         pairs:
           # AWS to AWS
-          - aws:us-east-1 aws:us-east-1
-          - aws:us-east-2 aws:us-west-2
+          - aws:us-east-1 aws:us-east-1 --multipart
+          - aws:us-east-2 aws:us-west-2 --multipart
           # GCP to GCP
-          - gcp:us-central1-a gcp:us-central1-a
-          - gcp:us-west1-a gcp:us-east1-a
+          - gcp:us-central1-a gcp:us-central1-a --multipart
+          - gcp:us-west1-a gcp:us-east1-a --multipart
           # Azure to Azure
-          #- azure:westus azure:westus
-          #- azure:eastus azure:westus
+          - azure:westus azure:westus
+          - azure:eastus azure:westus
           # cross cloud tests
-          - aws:us-west-1 gcp:us-west2-a
-          - gcp:us-west2-a aws:us-west-1
-          #- aws:us-west-1 azure:westus
-          #- azure:westus aws:us-west-1
-          #- gcp:us-west2-a azure:westus
-          #- azure:westus gcp:us-west2-a
+          - aws:us-west-1 gcp:us-west2-a --multipart
+          - gcp:us-west2-a aws:us-west-1 --multipart
+          - aws:us-west-1 azure:westus
+          - azure:westus aws:us-west-1
+          - gcp:us-west2-a azure:westus
+          - azure:westus gcp:us-west2-a
     timeout-minutes: 40
     env:
       STRATEGY_UUID: itest-${{ github.run_id }}-${{ github.run_attempt }}-${{ strategy.job-index }}

skyplane/api/dataplane.py

@@ -226,7 +226,7 @@ def copy_gateway_logs(self):
         # copy logs from all gateways in parallel
         def copy_log(instance):
             out_file = self.transfer_dir / f"gateway_{instance.uuid()}.stdout"
-            err_file = self.transfer_dir / f"gateway_{instance.uuid()}.stdout"
+            err_file = self.transfer_dir / f"gateway_{instance.uuid()}.stderr"
             logger.fs.info(f"[Dataplane.copy_gateway_logs] Copying logs from {instance.uuid()}: {out_file}")
             instance.run_command("sudo docker logs -t skyplane_gateway 2> /tmp/gateway.stderr > /tmp/gateway.stdout")
             instance.download_file("/tmp/gateway.stdout", out_file)

skyplane/api/tracker.py

@@ -1,6 +1,5 @@
 import functools
 from pprint import pprint
-from pprint import pprint
 import json
 import time
 from abc import ABC
@@ -222,7 +221,12 @@ def monitor_single_dst_helper(dst_region):
                 job.finalize()
         except Exception as e:
             UsageClient.log_exception(
-                "finalize job", e, args, self.dataplane.src_region_tag, self.dataplane.dst_region_tag, session_start_timestamp_ms
+                "finalize job",
+                e,
+                args,
+                self.dataplane.topology.src_region_tag,
+                self.dataplane.topology.dest_region_tags[0],
+                session_start_timestamp_ms,
             )
             raise e
         end_time = int(time.time())
@@ -308,7 +312,6 @@ def monitor_transfer(pd, self, region_tag):
                 self.job_pending_chunk_ids[job_uuid][region_tag] = self.job_pending_chunk_ids[job_uuid][region_tag].difference(
                     job_complete_chunk_ids
                 )
-
             # sleep
             time.sleep(0.05)
 
@@ -362,6 +365,7 @@ def query_bytes_remaining(self, region_tag: Optional[str] = None):
                 ]
             )
         logger.fs.debug(f"[TransferProgressTracker] Bytes remaining per job: {bytes_remaining_per_job}")
+        print(f"[TransferProgressTracker] Bytes remaining per job: {bytes_remaining_per_job}")
         return sum(bytes_remaining_per_job.values())
 
     def query_bytes_dispatched(self):