enable github actions on roblutt/windows-ci

.github/workflows/e2e_test.yml → .github/workflows/release_integration_canary.yml

@@ -1,23 +1,28 @@
-name: E2E Tests
+name: Release Integration Canary
 
 on:
-  workflow_dispatch:
+  schedule:
+    - cron: '0 */4 * * *'
 
 jobs:
-  E2E:
+  ReleaseIntegrationCanary:
     runs-on: ubuntu-latest
-    environment: mainline
+    environment: release
     permissions:
       id-token: write
+      contents: read
     steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: release
       - name: Configure AWS credentials
         uses: aws-actions/configure-aws-credentials@v4
         with:
-          role-to-assume: ${{ secrets.AWS_CODEBUILD_MAINLINE_E2E_ROLE }}
+          role-to-assume: ${{ secrets.AWS_CODEBUILD_RELEASE_INTEG_ROLE }}
           aws-region: us-west-2
           mask-aws-account-id: true
       - name: Run CodeBuild
         uses: aws-actions/aws-codebuild-run-build@v1
         with:
-          project-name: deadline-cloud-worker-agent-dev-E2ETest
+          project-name: deadline-cloud-worker-agent-IntegTest
           hide-cloudwatch-logs: true

.github/workflows/reuse_python_build.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       matrix:
         python-version: ['3.9', '3.10', '3.11']
-        os: [ubuntu-latest, macOS-latest]
+        os: [ubuntu-latest, macOS-latest, windows-latest]
     env:
       PYTHON: ${{ matrix.python-version }}
       CODEARTIFACT_REGION: "us-west-2"
@@ -45,19 +45,28 @@ jobs:
         aws-region: us-west-2
         mask-aws-account-id: true
 
-    - name: Install Hatch
+    - name: Install Hatch Posix
+      if: ${{ matrix.os == 'ubuntu-latest' || matrix.os == 'macOS-latest'}}
       shell: bash
       run: |
         CODEARTIFACT_AUTH_TOKEN=$(aws codeartifact get-authorization-token --domain ${{ secrets.CODEARTIFACT_DOMAIN }} --domain-owner ${{ secrets.CODEARTIFACT_ACCOUNT_ID }} --query authorizationToken --output text --region us-west-2)
         echo "::add-mask::$CODEARTIFACT_AUTH_TOKEN"
         echo CODEARTIFACT_AUTH_TOKEN=$CODEARTIFACT_AUTH_TOKEN >> $GITHUB_ENV
         pip install --upgrade hatch
 
+    - name: Install Hatch Windows
+      if: ${{ matrix.os == 'windows-latest'}}
+      run: |
+        $CODEARTIFACT_AUTH_TOKEN=$(aws codeartifact get-authorization-token --domain ${{ secrets.CODEARTIFACT_DOMAIN }} --domain-owner ${{ secrets.CODEARTIFACT_ACCOUNT_ID }} --query authorizationToken --output text --region us-west-2)
+        echo "::add-mask::$CODEARTIFACT_AUTH_TOKEN"
+        echo CODEARTIFACT_AUTH_TOKEN=$CODEARTIFACT_AUTH_TOKEN >> $env:GITHUB_ENV
+        pip install --upgrade hatch
+
     - name: Run Linting
       run: hatch run lint
 
     - name: Run Build
       run: hatch build
 
     - name: Run Tests
-      run: hatch run test
+      run: hatch run test

hatch.toml

@@ -40,12 +40,7 @@ PIP_INDEX_URL="https://aws:{env:CODEARTIFACT_AUTH_TOKEN}@{env:CODEARTIFACT_DOMAI
 
 [envs.codebuild.scripts]
 build = "hatch build"
+integ-test = "pytest --no-cov test/integ {args}"
 
 [envs.codebuild.env-vars]
-PIP_INDEX_URL=""
-
-[envs.integ.scripts]
-test = "pytest --no-cov {args:test/integ}"
-
-[envs.e2e.scripts]
-test = "pytest --no-cov {args:test/e2e}"
+PIP_INDEX_URL=""

pipeline/integ.sh

@@ -4,4 +4,4 @@ set -e
 
 pip install --upgrade pip
 pip install --upgrade hatch
-hatch run integ:test
+hatch run codebuild:integ-test

pyproject.toml

@@ -8,7 +8,7 @@ dynamic = ["version"]
 dependencies = [
     "requests ~= 2.29",
     "boto3 ~= 1.26",
-    "deadline == 0.29.*",
+    "deadline == 0.30.*",
     "openjd-sessions == 0.2.*",
     # tomli became tomllib in standard library in Python 3.11
     "tomli >= 1.1.0 ; python_version<'3.11'",
@@ -157,4 +157,4 @@ source = [ "src/" ]
 
 [tool.coverage.report]
 show_missing = true
-fail_under = 78
+fail_under = 69

src/deadline_worker_agent/api_models.py

@@ -328,7 +328,7 @@ class Attachments(TypedDict):
     manifests: list[ManifestProperties]
     """A list of all manifests and their configuration"""
 
-    assetLoadingMethod: NotRequired[str]
+    fileSystem: NotRequired[str]
     """Method to use when loading assets required for a job"""
 
 

src/deadline_worker_agent/aws/deadline/__init__.py

@@ -5,14 +5,16 @@
 from typing import Any, Optional
 from threading import Event
 from dataclasses import dataclass
+import random
 
 from botocore.retries.standard import RetryContext
 from botocore.exceptions import ClientError
 
 from deadline.client.api import get_telemetry_client, TelemetryClient
 
 from ..._version import __version__ as version  # noqa
-from ...startup.config import Configuration, Capabilities
+from ...startup.config import Configuration
+from ...startup.capabilities import Capabilities
 from ...boto import DeadlineClient, NoOverflowExponentialBackoff as Backoff
 from ...api_models import (
     AssumeFleetRoleForWorkerResponse,
@@ -120,6 +122,18 @@ def _get_error_reason_from_header(response: dict[str, Any]) -> Optional[str]:
     return response.get("reason", None)
 
 
+def _get_retry_after_seconds_from_header(response: dict[str, Any]) -> Optional[int]:
+    return response.get("retryAfterSeconds", None)
+
+
+def _apply_lower_bound_to_delay(delay: float, lower_bound: Optional[float] = None) -> float:
+    if lower_bound is not None and delay < lower_bound:
+        # We add just a tiny bit of jitter (20%) to the lower bound to reduce the probability
+        # of a group of workers all retry-storming in lock-step.
+        delay = lower_bound + random.uniform(0, 0.2 * lower_bound)
+    return delay
+
+
 def _get_resource_id_and_status_from_conflictexception_header(
     response: dict[str, Any]
 ) -> tuple[Optional[str], Optional[str]]:
@@ -155,7 +169,10 @@ def assume_fleet_role_for_worker(
             # Retry:
             #   ThrottlingException, InternalServerException
             delay = backoff.delay_amount(RetryContext(retry))
-            code = e.response.get("Error", {}).get("Code", None)
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
+            code = _get_error_code_from_header(e.response)
             if code == "ThrottlingException":
                 _logger.info(
                     f"Throttled while attempting to refresh Worker AWS Credentials. Retrying in {delay} seconds..."
@@ -216,12 +233,11 @@ def assume_queue_role_for_worker(
     retry = 0
     query_start_time = monotonic()
 
-    _logger.info("")
     # Note: Frozen credentials could expire while doing a retry loop; that's
     #  probably going to manifest as AccessDenied, but I'm not 100% certain.
     while True:
         if interrupt_event and interrupt_event.is_set():
-            raise DeadlineRequestInterrupted("GetQueueIamCredentials interrupted")
+            raise DeadlineRequestInterrupted("AssumeQueueRoleForWorker interrupted")
         try:
             response = deadline_client.assume_queue_role_for_worker(
                 farmId=farm_id, fleetId=fleet_id, workerId=worker_id, queueId=queue_id
@@ -235,7 +251,10 @@ def assume_queue_role_for_worker(
             # Retry:
             #   ThrottlingException, InternalServerException
             delay = backoff.delay_amount(RetryContext(retry))
-            code = e.response.get("Error", {}).get("Code", None)
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
+            code = _get_error_code_from_header(e.response)
             if code == "ThrottlingException":
                 _logger.info(
                     f"Throttled while attempting to refresh Worker AWS Credentials. Retrying in {delay} seconds..."
@@ -333,7 +352,10 @@ def batch_get_job_entity(
             # Retry:
             #   ThrottlingException, InternalServerException
             delay = backoff.delay_amount(RetryContext(retry))
-            code = e.response.get("Error", {}).get("Code", None)
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
+            code = _get_error_code_from_header(e.response)
             if code == "ThrottlingException":
                 _logger.info(f"Throttled calling BatchGetJobEntity. Retrying in {delay} seconds...")
             elif code == "InternalServerException":
@@ -377,6 +399,9 @@ def create_worker(
             break
         except ClientError as e:
             delay = backoff.delay_amount(RetryContext(retry))
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
             code = _get_error_code_from_header(e.response)
             if code == "ThrottlingException":
                 _logger.info(f"CreateWorker throttled. Retrying in {delay} seconds...")
@@ -444,6 +469,9 @@ def delete_worker(
             break
         except ClientError as e:
             delay = backoff.delay_amount(RetryContext(retry))
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
             code = _get_error_code_from_header(e.response)
             if code == "ThrottlingException":
                 _logger.info(f"DeleteWorker throttled. Retrying in {delay} seconds...")
@@ -487,16 +515,20 @@ def delete_worker(
 def update_worker(
     *,
     deadline_client: DeadlineClient,
-    config: Configuration,
+    farm_id: str,
+    fleet_id: str,
     worker_id: str,
     status: WorkerStatus,
+    capabilities: Optional[Capabilities] = None,
     host_properties: Optional[HostProperties] = None,
+    interrupt_event: Optional[Event] = None,
 ) -> UpdateWorkerResponse:
     """Calls the UpdateWorker API to update this Worker's status, capabilities, and/or host properties with the service.
 
     Raises:
        DeadlineRequestConditionallyRecoverableError
        DeadlineRequestUnrecoverableError
+       DeadlineRequestInterrupted
     """
 
     # Retry API call when being throttled
@@ -506,26 +538,32 @@ def update_worker(
     _logger.info(f"Invoking UpdateWorker to set {worker_id} to status={status.value}.")
 
     request: dict[str, Any] = dict(
-        farmId=config.farm_id,
-        fleetId=config.fleet_id,
+        farmId=farm_id,
+        fleetId=fleet_id,
         workerId=worker_id,
-        capabilities=config.capabilities.for_update_worker(),
         status=status.value,
     )
+    if capabilities:
+        request["capabilities"] = capabilities.for_update_worker()
     if host_properties:
         request["hostProperties"] = host_properties
 
+    _logger.debug("UpdateWorker request: %s", request)
     while True:
         # If true, then we're trying to go to STARTED but have determined that we must first
         # go to STOPPED
         must_stop_first = False
 
-        _logger.debug("UpdateWorker request: %s", request)
+        if interrupt_event and interrupt_event.is_set():
+            raise DeadlineRequestInterrupted("UpdateWorker interrupted")
         try:
             response = deadline_client.update_worker(**request)
             break
         except ClientError as e:
             delay = backoff.delay_amount(RetryContext(retry))
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
             code = _get_error_code_from_header(e.response)
 
             skip_sleep = False
@@ -578,7 +616,10 @@ def update_worker(
                 raise DeadlineRequestUnrecoverableError(e)
 
             if not skip_sleep:
-                sleep(delay)
+                if interrupt_event:
+                    interrupt_event.wait(delay)
+                else:
+                    sleep(delay)
                 retry += 1
         except Exception as e:
             _logger.error("Failed to start worker %s", worker_id)
@@ -589,9 +630,11 @@ def update_worker(
             try:
                 update_worker(
                     deadline_client=deadline_client,
-                    config=config,
+                    farm_id=farm_id,
+                    fleet_id=fleet_id,
                     worker_id=worker_id,
                     status=WorkerStatus.STOPPED,
+                    capabilities=capabilities,
                     host_properties=host_properties,
                 )
             except Exception:
@@ -695,6 +738,9 @@ def update_worker_schedule(
             break
         except ClientError as e:
             delay = backoff.delay_amount(RetryContext(retry))
+            delay = _apply_lower_bound_to_delay(
+                delay, _get_retry_after_seconds_from_header(e.response)
+            )
             code = _get_error_code_from_header(e.response)
 
             if code == "ThrottlingException":

src/deadline_worker_agent/boto/shim.py

@@ -281,7 +281,7 @@ def batch_get_job_entity(
                             "jobId": "job-21432d89b44a46cbaaeb2f1d5254e548",
                             "attachments": {
                                 "manifests": [],
-                                "assetLoadingMethod": "PRELOAD",
+                                "fileSystem": "COPIED",
                             },
                         },
                     },