diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/.gitignore b/samples/scenarios/WorkItemFilteringSplitActivities/.gitignore new file mode 100644 index 0000000..141c526 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/.gitignore @@ -0,0 +1,51 @@ +## Ignore Visual Studio temporary files, build results, and +## files generated by popular Visual Studio add-ons. +## +## Get latest from https://github.com/github/gitignore/blob/main/VisualStudio.gitignore + +# User-specific files +*.rsuser +*.suo +*.user +*.userosscache +*.sln.docstates + +# User-specific files (MonoDevelop/Xamarin Studio) +*.userprefs + +# Mono auto generated files +mono_crash.* + +# Build results +[Dd]ebug/ +[Rr]elease/ +x64/ +x86/ +[Ww][Ii][Nn]32/ +[Aa][Rr][Mm]/ +[Aa][Rr][Mm]64/ +bld/ +[Bb]in/ +[Oo]bj/ +[Ll]og/ +[Ll]ogs/ + +# Visual Studio 2015/2017 cache/options directory +.vs/ + +# NuGet Packages +*.nupkg +# NuGet Symbol Packages +*.snupkg +# The packages folder can be ignored because of Package Restore +**/[Pp]ackages/* +# except build/, which is used as an MSBuild target. +!**/[Pp]ackages/build/ +# Uncomment if necessary however generally it will be regenerated when needed +#!**/[Pp]ackages/repositories.config +# NuGet v3's project.json files produces more ignorable files +*.nuget.props +*.nuget.targets + +# Azure +.azure diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/README.md b/samples/scenarios/WorkItemFilteringSplitActivities/README.md new file mode 100644 index 0000000..e3269cc --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/README.md @@ -0,0 +1,288 @@ +# Work Item Filtering — Split Activities Sample + +This sample demonstrates **Work Item Filtering**, a feature that allows workers to declare which orchestrations, activities, and entities they can process. The Durable Task Scheduler (DTS) backend routes work items only to workers whose filters match, preventing workers from receiving work they cannot handle. + +Before work item filtering, all orchestrations, activities, and entities were handed to any connected worker regardless of what it actually hosted. This caused errors (or silent hangs) when a worker received a work item it didn't implement — especially problematic in multi-service deployments, rolling upgrades, and microservice topologies. With filtering, each worker registers its task set; DTS creates per-filter queues and routes work items to matching workers. If no filter is specified, behavior falls back to the "generic queue" (all workers receive everything). + +## Architecture + +``` +┌─────────────────────────────────────────────────────────────┐ +│ Durable Task Scheduler (DTS) │ +│ │ +│ Orchestration queue ──► routed to Orchestrator Worker only │ +│ ValidateOrder queue ──► routed to Validator Worker only │ +│ ShipOrder queue ──► routed to Shipper Worker only │ +└────────────┬──────────────────┬──────────────────┬──────────┘ + │ │ │ + ┌───────▼───────┐ ┌──────▼───────┐ ┌──────▼───────┐ + │ Orchestrator │ │ Validator │ │ Shipper │ + │ Worker │ │ Worker │ │ Worker │ + │ │ │ │ │ │ + │ Registers: │ │ Registers: │ │ Registers: │ + │ • OrderProc- │ │ • Validate- │ │ • ShipOrder │ + │ essing- │ │ Order │ │ │ + │ Orchestration│ │ │ │ │ + └───────────────┘ └───────────────┘ └───────────────┘ + + ┌───────────────┐ + │ Client │ + │ (Driver) │ + │ │ + │ Schedules new │ + │ orchestrations │ + │ and prints │ + │ results │ + └───────────────┘ +``` + +**Orchestrator Worker** runs orchestrations only — it has no activities registered. +**Validator Worker** runs `ValidateOrder` only — it has no orchestrations or other activities. +**Shipper Worker** runs `ShipOrder` only — same isolation. +**Client** schedules orchestrations and polls for completion. + +## The Orchestration + +`OrderProcessingOrchestration` performs two sequential activity calls: + +1. `ValidateOrder(orderId)` → routed to Validator Worker +2. `ShipOrder(orderId)` → routed to Shipper Worker + +Returns a combined result string. + +## Prerequisites + +- [.NET 10 SDK](https://dotnet.microsoft.com/download) (or later) +- [Docker](https://docs.docker.com/get-docker/) (for the DTS emulator) + +## Running Locally + +### 1. Start the DTS Emulator + +```bash +docker pull mcr.microsoft.com/durable-task/emulator:latest +docker run -d --name dts-emulator -p 8080:8080 -p 8082:8082 mcr.microsoft.com/durable-task/emulator:latest +``` + +The emulator dashboard is available at `http://localhost:8082`. + +### 2. Build all projects + +```bash +cd samples/scenarios/WorkItemFilteringSplitActivities +dotnet build +``` + +### 3. Start the three workers (each in a separate terminal) + +**Terminal 1 — Orchestrator Worker:** +```bash +dotnet run --project src/OrchestratorWorker +``` + +**Terminal 2 — Validator Worker (ValidateOrder activity):** +```bash +dotnet run --project src/ValidatorWorker +``` + +**Terminal 3 — Shipper Worker (ShipOrder activity):** +```bash +dotnet run --project src/ShipperWorker +``` + +### 4. Run the Client (in a fourth terminal) + +```bash +dotnet run --project src/Client +``` + +## Expected Output + +The client runs in a **continuous loop**, scheduling a batch of 3 orchestrations every 30 seconds for 10 minutes. This makes it easy to observe scaling behavior over time. + +### Client terminal + +``` +10:30:01 [Client] === Work Item Filtering Demo — Client === +10:30:01 [Client] Will schedule 3 orchestrations every 30s for 10 minutes. + +10:30:01 [Client] --- Batch #1 at 10:30:01 --- +10:30:01 [Client] Scheduling orchestration #1 with orderId='ORD-B001-001'... +10:30:01 [Client] -> Scheduled with InstanceId=abc123 +10:30:01 [Client] Scheduling orchestration #2 with orderId='ORD-B001-002'... +10:30:01 [Client] -> Scheduled with InstanceId=def456 +10:30:01 [Client] Scheduling orchestration #3 with orderId='ORD-B001-003'... +10:30:01 [Client] -> Scheduled with InstanceId=ghi789 +10:30:01 [Client] All 3 orchestrations scheduled. Waiting for completion... +10:30:02 [Client] COMPLETED | InstanceId=abc123 | Output: Order 'ORD-B001-001' => Validation: [Order ORD-B001-001 is valid], Shipping: [Shipped with tracking TRACK-ORD-B001-001-4271] +... +10:30:02 [Client] === RESULTS: 3 completed, 0 failed, 3 total === +10:30:02 [Client] Sleeping 30s until next batch... + +10:30:32 [Client] --- Batch #2 at 10:30:32 --- +... +``` + +### Orchestrator Worker terminal (orchestrations only — no activities) + +``` +10:30:02 [Orchestrator] Orchestration | Name=OrderProcessingOrchestration | InstanceId=abc123 | Processing order 'ORD-B001-001' +10:30:02 [Orchestrator] Orchestration | InstanceId=abc123 | Dispatching ValidateOrder to Validator Worker... +10:30:02 [Orchestrator] Orchestration | InstanceId=abc123 | Dispatching ShipOrder to Shipper Worker... +10:30:02 [Orchestrator] Orchestration | InstanceId=abc123 | Completed: Order 'ORD-B001-001' => Validation: [...], Shipping: [...] +``` + +### Validator Worker terminal (ValidateOrder only — no ShipOrder, no orchestrations) + +``` +10:30:02 [Validator] Activity | Name=ValidateOrder | InstanceId=abc123 | Validating order 'ORD-B001-001'... +10:30:02 [Validator] Activity | Name=ValidateOrder | InstanceId=abc123 | Result: Order ORD-B001-001 is valid +10:30:02 [Validator] Activity | Name=ValidateOrder | InstanceId=def456 | Validating order 'ORD-B001-002'... +``` + +### Shipper Worker terminal (ShipOrder only — no ValidateOrder, no orchestrations) + +``` +10:30:02 [Shipper] Activity | Name=ShipOrder | InstanceId=abc123 | Shipping order 'ORD-B001-001'... +10:30:02 [Shipper] Activity | Name=ShipOrder | InstanceId=abc123 | Result: Shipped with tracking TRACK-ORD-B001-001-4271 +10:30:02 [Shipper] Activity | Name=ShipOrder | InstanceId=def456 | Shipping order 'ORD-B001-002'... +``` + +**Key observation:** Each worker processes **only** its registered work item types. No cross-processing occurs. + +## What to Try Next: Strict Routing Experiment + +1. **Stop Shipper Worker** (Ctrl+C in Terminal 3). +2. Run the Client again to schedule new orchestrations. +3. Observe that: + - Orchestrator Worker picks up and starts orchestrations. + - Validator Worker completes `ValidateOrder` for each order. + - `ShipOrder` work items **remain pending** — they are not delivered to Validator Worker or Orchestrator Worker. + - The orchestrations stay in "Running" status, waiting for the `ShipOrder` activity to complete. +4. **Restart Shipper Worker** — the pending `ShipOrder` work items are immediately delivered and the orchestrations complete. + +This demonstrates that filtering is strict: work items are routed only to workers with matching filters. There is no fallback to other workers. + +## How It Works + +Each worker process registers its tasks in a `DurableTaskRegistry` via `AddAllGeneratedTasks()` (which picks up classes decorated with `[DurableTask]`). When the worker connects to DTS, the SDK automatically constructs **work item filters** from the registry: + +- Orchestrator Worker's filter: `orchestrations: [OrderProcessingOrchestration]` +- Validator Worker's filter: `activities: [ValidateOrder]` +- Shipper Worker's filter: `activities: [ShipOrder]` + +DTS creates per-filter queues and routes each work item to the matching queue. If a filter list is empty for a given type (e.g., Validator Worker has no orchestration filter), that worker simply never receives work items of that type. + +This is all **automatic** — no explicit `UseWorkItemFilters()` call is needed. The SDK generates filters from whatever you register. To override or opt out, you can use `UseWorkItemFilters(customFilters)` or `UseWorkItemFilters(null)` respectively. + +## Deploying to Azure + +This sample includes full infrastructure-as-code (Bicep) and an `azure.yaml` for one-command deployment via [Azure Developer CLI (`azd`)](https://learn.microsoft.com/azure/developer/azure-developer-cli/). + +### What Gets Deployed + +| Resource | Purpose | +|---|---| +| **Resource Group** | Contains all resources | +| **Durable Task Scheduler** (Consumption SKU) | Managed orchestration backend | +| **Task Hub** | Logical unit for orchestrations and work items | +| **Container Apps Environment** | Shared hosting environment with VNet integration | +| **Azure Container Registry** | Stores Docker images for each service | +| **User-Assigned Managed Identity** | Shared identity with DTS Worker/Client RBAC role | +| **4 Container Apps** | Client, Orchestrator Worker, Validator Worker, Shipper Worker | + +### Deploy with `azd` + +```bash +cd samples/scenarios/WorkItemFilteringSplitActivities +azd up +``` + +You'll be prompted for an environment name, subscription, and location. The deployment takes ~5 minutes. + +### KEDA Scaling with DTS + +Each worker Container App is configured with a **DTS-aware KEDA custom scale rule** (`azure-durabletask-scheduler`) that scales based on the **work item backlog** in the task hub. The key parameter is `workItemType`, which tells the scaler what kind of work to monitor: + +| Container App | Service Name | `workItemType` | Scales on | +|---|---|---|---| +| **Client** | `client` | `Orchestration` | Pending orchestration work items | +| **Orchestrator Worker** | `orchestrator-worker` | `Orchestration` | Pending orchestration work items | +| **Validator Worker** | `validator-worker` | `Activity` | Pending activity work items | +| **Shipper Worker** | `shipper-worker` | `Activity` | Pending activity work items | + +The scale rule metadata (from [app.bicep](infra/app/app.bicep)): + +```bicep +scaleRuleType: 'azure-durabletask-scheduler' +scaleRuleMetadata: { + endpoint: dtsEndpoint // DTS scheduler URL + maxConcurrentWorkItemsCount: '1' + taskhubName: taskHubName + workItemType: workItemType // 'Orchestration' or 'Activity' +} +scaleRuleIdentity: userAssignedManagedIdentity.resourceId +``` + +- Workers scale from **0 to 10** replicas. When the client finishes its loop and no more work items arrive, workers scale back to zero. +- The `scaleRuleIdentity` uses the shared user-assigned managed identity to authenticate with DTS, so no connection strings or secrets are needed for scaling. +- `maxConcurrentWorkItemsCount: '1'` means KEDA will scale up one replica per pending work item, up to the max. + +### Manual Deployment (without `azd`) + +Set the `ENDPOINT` and `TASKHUB` environment variables to point to your deployed scheduler: + +```bash +export ENDPOINT="https://your-scheduler.westus2.durabletask.io" +export TASKHUB="your-taskhub-name" +``` + +The workers and client will automatically use `DefaultAzureCredential` for authentication. Make sure the identity running each process has the **Durable Task Scheduler Worker** / **Durable Task Scheduler Client** role on the scheduler resource. + +## Project Structure + +``` +WorkItemFilteringSplitActivities/ +├── WorkItemFilteringSplitActivities.sln +├── README.md +├── azure.yaml # azd service definitions +├── .gitignore +├── infra/ # Bicep infrastructure-as-code +│ ├── main.bicep # Top-level — resource group, DTS, container apps +│ ├── main.parameters.json +│ ├── abbreviations.json +│ ├── app/ +│ │ ├── app.bicep # Per-service container app (with KEDA scale rule) +│ │ ├── dts.bicep # DTS scheduler + task hub +│ │ └── user-assigned-identity.bicep +│ └── core/ +│ ├── host/ # Container Apps Environment, Registry, App template +│ ├── networking/ # VNet +│ └── security/ # ACR pull role, DTS role assignments +└── src/ + ├── Client/ # Schedules orchestrations in a loop, prints results + │ ├── Client.csproj + │ ├── Program.cs + │ └── Dockerfile + ├── OrchestratorWorker/ # Orchestrator Worker — runs orchestrations only + │ ├── OrchestratorWorker.csproj + │ ├── Program.cs + │ ├── OrderProcessingOrchestration.cs + │ └── Dockerfile + ├── ValidatorWorker/ # Validator Worker — runs ValidateOrder activity only + │ ├── ValidatorWorker.csproj + │ ├── Program.cs + │ ├── ValidateOrder.cs + │ └── Dockerfile + └── ShipperWorker/ # Shipper Worker — runs ShipOrder activity only + ├── ShipperWorker.csproj + ├── Program.cs + ├── ShipOrder.cs + └── Dockerfile +``` + +## Reference + +- [Work Item Filtering PR (durabletask-dotnet #616)](https://github.com/microsoft/durabletask-dotnet/pull/616) +- [Durable Task Scheduler documentation](https://learn.microsoft.com/azure/durable-task-scheduler/) +- [Durable Task .NET SDK](https://github.com/microsoft/durabletask-dotnet) diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/WorkItemFilteringSplitActivities.sln b/samples/scenarios/WorkItemFilteringSplitActivities/WorkItemFilteringSplitActivities.sln new file mode 100644 index 0000000..2a396d6 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/WorkItemFilteringSplitActivities.sln @@ -0,0 +1,42 @@ +Microsoft Visual Studio Solution File, Format Version 12.00 +# Visual Studio Version 17 +VisualStudioVersion = 17.5.2.0 +MinimumVisualStudioVersion = 10.0.40219.1 +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "OrchestratorWorker", "src\OrchestratorWorker\OrchestratorWorker.csproj", "{A1B2C3D4-E5F6-7890-ABCD-EF1234567890}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ValidatorWorker", "src\ValidatorWorker\ValidatorWorker.csproj", "{B2C3D4E5-F6A7-8901-BCDE-F12345678901}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ShipperWorker", "src\ShipperWorker\ShipperWorker.csproj", "{C3D4E5F6-A7B8-9012-CDEF-123456789012}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Client", "src\Client\Client.csproj", "{D4E5F6A7-B8C9-0123-DEF0-234567890123}" +EndProject +Global + GlobalSection(SolutionConfigurationPlatforms) = preSolution + Debug|Any CPU = Debug|Any CPU + Release|Any CPU = Release|Any CPU + EndGlobalSection + GlobalSection(ProjectConfigurationPlatforms) = postSolution + {A1B2C3D4-E5F6-7890-ABCD-EF1234567890}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {A1B2C3D4-E5F6-7890-ABCD-EF1234567890}.Debug|Any CPU.Build.0 = Debug|Any CPU + {A1B2C3D4-E5F6-7890-ABCD-EF1234567890}.Release|Any CPU.ActiveCfg = Release|Any CPU + {A1B2C3D4-E5F6-7890-ABCD-EF1234567890}.Release|Any CPU.Build.0 = Release|Any CPU + {B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Debug|Any CPU.Build.0 = Debug|Any CPU + {B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Release|Any CPU.ActiveCfg = Release|Any CPU + {B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Release|Any CPU.Build.0 = Release|Any CPU + {C3D4E5F6-A7B8-9012-CDEF-123456789012}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {C3D4E5F6-A7B8-9012-CDEF-123456789012}.Debug|Any CPU.Build.0 = Debug|Any CPU + {C3D4E5F6-A7B8-9012-CDEF-123456789012}.Release|Any CPU.ActiveCfg = Release|Any CPU + {C3D4E5F6-A7B8-9012-CDEF-123456789012}.Release|Any CPU.Build.0 = Release|Any CPU + {D4E5F6A7-B8C9-0123-DEF0-234567890123}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {D4E5F6A7-B8C9-0123-DEF0-234567890123}.Debug|Any CPU.Build.0 = Debug|Any CPU + {D4E5F6A7-B8C9-0123-DEF0-234567890123}.Release|Any CPU.ActiveCfg = Release|Any CPU + {D4E5F6A7-B8C9-0123-DEF0-234567890123}.Release|Any CPU.Build.0 = Release|Any CPU + EndGlobalSection + GlobalSection(SolutionProperties) = preSolution + HideSolutionNode = FALSE + EndGlobalSection + GlobalSection(ExtensibilityGlobals) = postSolution + SolutionGuid = {E5F6A7B8-C9D0-1234-EF01-345678901234} + EndGlobalSection +EndGlobal diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/azure.yaml b/samples/scenarios/WorkItemFilteringSplitActivities/azure.yaml new file mode 100644 index 0000000..6185ee6 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/azure.yaml @@ -0,0 +1,34 @@ +# yaml-language-server: $schema=https://raw.githubusercontent.com/Azure/azure-dev/main/schemas/v1.0/azure.yaml.json + +metadata: + template: work-item-filtering-split-activities-dotnet +name: work-item-filtering-split-activities +services: + client: + project: ./src/Client + language: csharp + host: containerapp + apiVersion: 2025-01-01 + docker: + path: ./Dockerfile + orchestrator-worker: + project: ./src/OrchestratorWorker + language: csharp + host: containerapp + apiVersion: 2025-01-01 + docker: + path: ./Dockerfile + validator-worker: + project: ./src/ValidatorWorker + language: csharp + host: containerapp + apiVersion: 2025-01-01 + docker: + path: ./Dockerfile + shipper-worker: + project: ./src/ShipperWorker + language: csharp + host: containerapp + apiVersion: 2025-01-01 + docker: + path: ./Dockerfile diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/abbreviations.json b/samples/scenarios/WorkItemFilteringSplitActivities/infra/abbreviations.json new file mode 100644 index 0000000..1f9a112 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/abbreviations.json @@ -0,0 +1,139 @@ +{ + "analysisServicesServers": "as", + "apiManagementService": "apim-", + "appConfigurationStores": "appcs-", + "appManagedEnvironments": "cae-", + "appContainerApps": "ca-", + "authorizationPolicyDefinitions": "policy-", + "automationAutomationAccounts": "aa-", + "blueprintBlueprints": "bp-", + "blueprintBlueprintsArtifacts": "bpa-", + "cacheRedis": "redis-", + "cdnProfiles": "cdnp-", + "cdnProfilesEndpoints": "cdne-", + "cognitiveServicesAccounts": "cog-", + "cognitiveServicesFormRecognizer": "cog-fr-", + "cognitiveServicesTextAnalytics": "cog-ta-", + "cognitiveServicesSpeech": "cog-sp-", + "computeAvailabilitySets": "avail-", + "computeCloudServices": "cld-", + "computeDiskEncryptionSets": "des", + "computeDisks": "disk", + "computeDisksOs": "osdisk", + "computeGalleries": "gal", + "computeSnapshots": "snap-", + "computeVirtualMachines": "vm", + "computeVirtualMachineScaleSets": "vmss-", + "containerInstanceContainerGroups": "ci", + "containerRegistryRegistries": "cr", + "containerServiceManagedClusters": "aks-", + "databricksWorkspaces": "dbw-", + "dataFactoryFactories": "adf-", + "dataLakeAnalyticsAccounts": "dla", + "dataLakeStoreAccounts": "dls", + "dataMigrationServices": "dms-", + "dBforMySQLServers": "mysql-", + "dBforPostgreSQLServers": "psql-", + "devicesIotHubs": "iot-", + "devicesProvisioningServices": "provs-", + "devicesProvisioningServicesCertificates": "pcert-", + "documentDBDatabaseAccounts": "cosmos-", + "eventGridDomains": "evgd-", + "eventGridDomainsTopics": "evgt-", + "eventGridEventSubscriptions": "evgs-", + "eventHubNamespaces": "evhns-", + "eventHubNamespacesEventHubs": "evh-", + "hdInsightClustersHadoop": "hadoop-", + "hdInsightClustersHbase": "hbase-", + "hdInsightClustersKafka": "kafka-", + "hdInsightClustersMl": "mls-", + "hdInsightClustersSpark": "spark-", + "hdInsightClustersStorm": "storm-", + "hybridComputeMachines": "arcs-", + "insightsActionGroups": "ag-", + "insightsComponents": "appi-", + "keyVaultVaults": "kv-", + "kubernetesConnectedClusters": "arck", + "kustoClusters": "dec", + "kustoClustersDatabases": "dedb", + "loadTesting": "lt-", + "logicIntegrationAccounts": "ia-", + "logicWorkflows": "logic-", + "machineLearningServicesWorkspaces": "mlw-", + "managedIdentityUserAssignedIdentities": "id-", + "managementManagementGroups": "mg-", + "migrateAssessmentProjects": "migr-", + "networkApplicationGateways": "agw-", + "networkApplicationSecurityGroups": "asg-", + "networkAzureFirewalls": "afw-", + "networkBastionHosts": "bas-", + "networkConnections": "con-", + "networkDnsZones": "dnsz-", + "networkExpressRouteCircuits": "erc-", + "networkFirewallPolicies": "afwp-", + "networkFirewallPoliciesWebApplication": "waf", + "networkFirewallPoliciesRuleGroups": "wafrg", + "networkFrontDoors": "fd-", + "networkFrontdoorWebApplicationFirewallPolicies": "fdfp-", + "networkLoadBalancersExternal": "lbe-", + "networkLoadBalancersInternal": "lbi-", + "networkLoadBalancersInboundNatRules": "rule-", + "networkLocalNetworkGateways": "lgw-", + "networkNatGateways": "ng-", + "networkNetworkInterfaces": "nic-", + "networkNetworkSecurityGroups": "nsg-", + "networkNetworkSecurityGroupsSecurityRules": "nsgsr-", + "networkNetworkWatchers": "nw-", + "networkPrivateDnsZones": "pdnsz-", + "networkPrivateLinkServices": "pl-", + "networkPublicIPAddresses": "pip-", + "networkPublicIPPrefixes": "ippre-", + "networkRouteFilters": "rf-", + "networkRouteTables": "rt-", + "networkRouteTablesRoutes": "udr-", + "networkTrafficManagerProfiles": "traf-", + "networkVirtualNetworkGateways": "vgw-", + "networkVirtualNetworks": "vnet-", + "networkVirtualNetworksSubnets": "snet-", + "networkVirtualNetworksVirtualNetworkPeerings": "peer-", + "networkVirtualWans": "vwan-", + "networkVpnGateways": "vpng-", + "networkVpnGatewaysVpnConnections": "vcn-", + "networkVpnGatewaysVpnSites": "vst-", + "notificationHubsNamespaces": "ntfns-", + "notificationHubsNamespacesNotificationHubs": "ntf-", + "operationalInsightsWorkspaces": "log-", + "portalDashboards": "dash-", + "powerBIDedicatedCapacities": "pbi-", + "purviewAccounts": "pview-", + "recoveryServicesVaults": "rsv-", + "resourcesResourceGroups": "rg-", + "searchSearchServices": "srch-", + "serviceBusNamespaces": "sb-", + "serviceBusNamespacesQueues": "sbq-", + "serviceBusNamespacesTopics": "sbt-", + "serviceEndPointPolicies": "se-", + "serviceFabricClusters": "sf-", + "signalRServiceSignalR": "sigr", + "sqlManagedInstances": "sqlmi-", + "sqlServers": "sql-", + "sqlServersDataWarehouse": "sqldw-", + "sqlServersDatabases": "sqldb-", + "sqlServersDatabasesStretch": "sqlstrdb-", + "storageStorageAccounts": "st", + "storageStorageAccountsVm": "stvm", + "storSimpleManagers": "ssimp", + "streamAnalyticsCluster": "asa-", + "synapseWorkspaces": "syn", + "synapseWorkspacesAnalyticsWorkspaces": "synw", + "synapseWorkspacesSqlPoolsDedicated": "syndp", + "synapseWorkspacesSqlPoolsSpark": "synsp", + "timeSeriesInsightsEnvironments": "tsi-", + "webServerFarms": "plan-", + "webSitesAppService": "app-", + "webSitesAppServiceEnvironment": "ase-", + "webSitesFunctions": "func-", + "webStaticSites": "stapp-", + "dts": "dts-", + "taskhub": "taskhub-" +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/app.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/app.bicep new file mode 100644 index 0000000..ec4953a --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/app.bicep @@ -0,0 +1,66 @@ +param appName string +param location string = resourceGroup().location +param tags object = {} + +param identityName string +param containerAppsEnvironmentName string +param containerRegistryName string +param serviceName string = 'aca' +param dtsEndpoint string +param taskHubName string + +@description('DTS work item type for KEDA scaling: Orchestration or Activity') +param workItemType string = 'Orchestration' + +type managedIdentity = { + resourceId: string + clientId: string +} + +@description('Unique identifier for user-assigned managed identity.') +param userAssignedManagedIdentity managedIdentity + +module containerAppsApp '../core/host/container-app.bicep' = { + name: 'container-apps-${serviceName}' + params: { + name: appName + containerAppsEnvironmentName: containerAppsEnvironmentName + containerRegistryName: containerRegistryName + location: location + tags: union(tags, { 'azd-service-name': serviceName }) + ingressEnabled: false + secrets: { + 'azure-managed-identity-client-id': userAssignedManagedIdentity.clientId + } + env: [ + { + name: 'AZURE_MANAGED_IDENTITY_CLIENT_ID' + secretRef: 'azure-managed-identity-client-id' + } + { + name: 'ENDPOINT' + value: dtsEndpoint + } + { + name: 'TASKHUB' + value: taskHubName + } + ] + identityName: identityName + containerMinReplicas: 0 + containerMaxReplicas: 10 + enableCustomScaleRule: true + scaleRuleName: 'dtsscaler-${serviceName}' + scaleRuleType: 'azure-durabletask-scheduler' + scaleRuleMetadata: { + endpoint: dtsEndpoint + maxConcurrentWorkItemsCount: '1' + taskhubName: taskHubName + workItemType: workItemType + } + scaleRuleIdentity: userAssignedManagedIdentity.resourceId + } +} + +output endpoint string = containerAppsApp.outputs.uri +output envName string = containerAppsApp.outputs.name diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/dts.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/dts.bicep new file mode 100644 index 0000000..6280e5a --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/dts.bicep @@ -0,0 +1,31 @@ +param ipAllowlist array +param location string +param tags object = {} +param name string +param taskhubname string +param skuName string +param skuCapacity int = 0 + +resource dts 'Microsoft.DurableTask/schedulers@2025-11-01' = { + location: location + tags: tags + name: name + properties: { + ipAllowlist: ipAllowlist + sku: skuName == 'Dedicated' ? { + name: skuName + capacity: skuCapacity + } : { + name: skuName + } + } +} + +resource taskhub 'Microsoft.DurableTask/schedulers/taskhubs@2025-11-01' = { + parent: dts + name: taskhubname +} + +output dts_NAME string = dts.name +output dts_URL string = dts.properties.endpoint +output TASKHUB_NAME string = taskhub.name diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/user-assigned-identity.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/user-assigned-identity.bicep new file mode 100644 index 0000000..0583ab8 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/app/user-assigned-identity.bicep @@ -0,0 +1,17 @@ +metadata description = 'Creates a Microsoft Entra user-assigned identity.' + +param name string +param location string = resourceGroup().location +param tags object = {} + +resource identity 'Microsoft.ManagedIdentity/userAssignedIdentities@2023-01-31' = { + name: name + location: location + tags: tags +} + +output name string = identity.name +output resourceId string = identity.id +output principalId string = identity.properties.principalId +output clientId string = identity.properties.clientId +output tenantId string = identity.properties.tenantId diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-app.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-app.bicep new file mode 100644 index 0000000..8b1df5c --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-app.bicep @@ -0,0 +1,195 @@ +metadata description = 'Creates a container app in an Azure Container App environment.' +param name string +param location string = resourceGroup().location +param tags object = {} + +@description('Allowed origins') +param allowedOrigins array = [] + +@description('Name of the environment for container apps') +param containerAppsEnvironmentName string + +@description('CPU cores allocated to a single container instance, e.g., 0.5') +param containerCpuCoreCount string = '0.5' + +@description('The maximum number of replicas to run. Must be at least 1.') +@minValue(1) +param containerMaxReplicas int = 10 + +@description('Memory allocated to a single container instance, e.g., 1Gi') +param containerMemory string = '1.0Gi' + +@description('The minimum number of replicas to run. Must be at least 0.') +param containerMinReplicas int = 0 + +@description('The name of the container') +param containerName string = 'main' + +@description('Enable custom scale rule') +param enableCustomScaleRule bool = false + +@description('Scale rule name') +param scaleRuleName string = 'scaler' + +@description('Scale rule type') +param scaleRuleType string = '' + +@description('Scale rule metadata') +param scaleRuleMetadata object = {} + +@description('Scale rule identity') +param scaleRuleIdentity string = '' + +@description('The name of the container registry') +param containerRegistryName string = '' + +@description('Hostname suffix for container registry. Set when deploying to sovereign clouds') +param containerRegistryHostSuffix string = 'azurecr.io' + +@description('The protocol used by Dapr to connect to the app, e.g., http or grpc') +@allowed([ 'http', 'grpc' ]) +param daprAppProtocol string = 'http' + +@description('The Dapr app ID') +param daprAppId string = containerName + +@description('Enable Dapr') +param daprEnabled bool = false + +@description('The environment variables for the container') +param env array = [] + +@description('Specifies if the resource ingress is exposed externally') +param external bool = true + +@description('The name of the user-assigned identity') +param identityName string = '' + +@description('The type of identity for the resource') +@allowed([ 'None', 'SystemAssigned', 'UserAssigned' ]) +param identityType string = 'None' + +@description('The name of the container image') +param imageName string = '' + +@description('Specifies if Ingress is enabled for the container app') +param ingressEnabled bool = true + +param revisionMode string = 'Single' + +@description('The secrets required for the container') +@secure() +param secrets object = {} + +@description('The service binds associated with the container') +param serviceBinds array = [] + +@description('The name of the container apps add-on to use. e.g. redis') +param serviceType string = '' + +@description('The target port for the container') +param targetPort int = 80 + +resource userIdentity 'Microsoft.ManagedIdentity/userAssignedIdentities@2023-01-31' existing = if (!empty(identityName)) { + name: identityName +} + +// Private registry support requires both an ACR name and a User Assigned managed identity +var usePrivateRegistry = !empty(identityName) && !empty(containerRegistryName) + +// Automatically set to `UserAssigned` when an `identityName` has been set +var normalizedIdentityType = !empty(identityName) ? 'UserAssigned' : identityType + +module containerRegistryAccess '../security/registry-access.bicep' = if (usePrivateRegistry) { + name: '${deployment().name}-registry-access' + params: { + containerRegistryName: containerRegistryName + principalId: usePrivateRegistry ? userIdentity!.properties.principalId : '' + } +} + +resource app 'Microsoft.App/containerApps@2025-01-01' = { + name: name + location: location + tags: tags + // It is critical that the identity is granted ACR pull access before the app is created + // otherwise the container app will throw a provision error + // This also forces us to use an user assigned managed identity since there would no way to + // provide the system assigned identity with the ACR pull access before the app is created + dependsOn: usePrivateRegistry ? [ containerRegistryAccess ] : [] + identity: { + type: normalizedIdentityType + userAssignedIdentities: !empty(identityName) && normalizedIdentityType == 'UserAssigned' ? { '${userIdentity.id}': {} } : null + } + properties: { + managedEnvironmentId: containerAppsEnvironment.id + configuration: { + activeRevisionsMode: revisionMode + ingress: ingressEnabled ? { + external: external + targetPort: targetPort + transport: 'auto' + corsPolicy: { + allowedOrigins: union([ 'https://portal.azure.com', 'https://ms.portal.azure.com' ], allowedOrigins) + } + } : null + dapr: daprEnabled ? { + enabled: true + appId: daprAppId + appProtocol: daprAppProtocol + appPort: ingressEnabled ? targetPort : 0 + } : { enabled: false } + secrets: [for secret in items(secrets): { + name: secret.key + #disable-next-line use-secure-value-for-secure-inputs + value: secret.value + }] + service: !empty(serviceType) ? { type: serviceType } : null + registries: usePrivateRegistry ? [ + { + server: '${containerRegistryName}.${containerRegistryHostSuffix}' + identity: userIdentity.id + } + ] : [] + } + template: { + serviceBinds: !empty(serviceBinds) ? serviceBinds : null + containers: [ + { + image: !empty(imageName) ? imageName : 'mcr.microsoft.com/azuredocs/containerapps-helloworld:latest' + name: containerName + env: env + resources: { + cpu: json(containerCpuCoreCount) + memory: containerMemory + } + } + ] + scale: { + minReplicas: containerMinReplicas + maxReplicas: containerMaxReplicas + rules: enableCustomScaleRule ? [ + { + name: scaleRuleName + custom: { + type: scaleRuleType + metadata: scaleRuleMetadata + identity: scaleRuleIdentity + } + } + ] : [] + } + } + } +} + +resource containerAppsEnvironment 'Microsoft.App/managedEnvironments@2023-05-01' existing = { + name: containerAppsEnvironmentName +} + +output defaultDomain string = containerAppsEnvironment.properties.defaultDomain +output identityPrincipalId string = normalizedIdentityType == 'None' ? '' : (empty(identityName) ? app.identity.principalId : userIdentity!.properties.principalId) +output imageName string = imageName +output name string = app.name +output serviceBind object = !empty(serviceType) ? { serviceId: app.id, name: name } : {} +output uri string = ingressEnabled ? 'https://${app.properties.configuration.ingress.fqdn}' : '' diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-apps-environment.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-apps-environment.bicep new file mode 100644 index 0000000..560d293 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-apps-environment.bicep @@ -0,0 +1,27 @@ +metadata description = 'Creates an Azure Container Apps environment.' +param name string +param location string = resourceGroup().location +param tags object = {} + +@description('Subnet resource ID for the Container Apps environment') +param subnetResourceId string = '' + +@description('Whether to use an internal or external load balancer') +@allowed(['Internal', 'External']) +param loadBalancerType string = 'External' + +resource containerAppsEnvironment 'Microsoft.App/managedEnvironments@2023-05-01' = { + name: name + location: location + tags: tags + properties: { + vnetConfiguration: !empty(subnetResourceId) ? { + infrastructureSubnetId: subnetResourceId + internal: loadBalancerType == 'Internal' + } : null + } +} + +output defaultDomain string = containerAppsEnvironment.properties.defaultDomain +output id string = containerAppsEnvironment.id +output name string = containerAppsEnvironment.name diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-apps.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-apps.bicep new file mode 100644 index 0000000..4511423 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-apps.bicep @@ -0,0 +1,44 @@ +metadata description = 'Creates an Azure Container Registry and an Azure Container Apps environment.' +param name string +param location string = resourceGroup().location +param tags object = {} + +param containerAppsEnvironmentName string +param containerRegistryName string +param containerRegistryAdminUserEnabled bool = false + +// Virtual network and subnet parameters +param subnetResourceId string = '' +param loadBalancerType string = 'External' + +module containerAppsEnvironment 'container-apps-environment.bicep' = { + name: '${name}-container-apps-environment' + params: { + name: containerAppsEnvironmentName + location: location + tags: tags + subnetResourceId: subnetResourceId + loadBalancerType: loadBalancerType + } +} + +module containerRegistry 'container-registry.bicep' = { + name: '${name}-container-registry' + params: { + name: containerRegistryName + location: location + adminUserEnabled: containerRegistryAdminUserEnabled + tags: tags + sku: { + name: 'Standard' + } + anonymousPullEnabled: false + } +} + +output defaultDomain string = containerAppsEnvironment.outputs.defaultDomain +output environmentName string = containerAppsEnvironment.outputs.name +output environmentId string = containerAppsEnvironment.outputs.id + +output registryLoginServer string = containerRegistry.outputs.loginServer +output registryName string = containerRegistry.outputs.name diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-registry.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-registry.bicep new file mode 100644 index 0000000..9ea04a2 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/host/container-registry.bicep @@ -0,0 +1,134 @@ +metadata description = 'Creates an Azure Container Registry.' +param name string +param location string = resourceGroup().location +param tags object = {} + +@description('Indicates whether admin user is enabled') +param adminUserEnabled bool = false + +@description('Indicates whether anonymous pull is enabled') +param anonymousPullEnabled bool = false + +@description('Azure ad authentication as arm policy settings') +param azureADAuthenticationAsArmPolicy object = { + status: 'enabled' +} + +@description('Indicates whether data endpoint is enabled') +param dataEndpointEnabled bool = false + +@description('Encryption settings') +param encryption object = { + status: 'disabled' +} + +@description('Export policy settings') +param exportPolicy object = { + status: 'enabled' +} + +@description('Metadata search settings') +param metadataSearch string = 'Disabled' + +@description('Options for bypassing network rules') +param networkRuleBypassOptions string = 'AzureServices' + +@description('Public network access setting') +param publicNetworkAccess string = 'Enabled' + +@description('Quarantine policy settings') +param quarantinePolicy object = { + status: 'disabled' +} + +@description('Retention policy settings') +param retentionPolicy object = { + days: 7 + status: 'disabled' +} + +@description('Scope maps setting') +param scopeMaps array = [] + +@description('SKU settings') +param sku object = { + name: 'Basic' +} + +@description('Soft delete policy settings') +param softDeletePolicy object = { + retentionDays: 7 + status: 'disabled' +} + +@description('Trust policy settings') +param trustPolicy object = { + type: 'Notary' + status: 'disabled' +} + +@description('Zone redundancy setting') +param zoneRedundancy string = 'Disabled' + +@description('The log analytics workspace ID used for logging and monitoring') +param workspaceId string = '' + +// 2023-11-01-preview needed for metadataSearch +resource containerRegistry 'Microsoft.ContainerRegistry/registries@2023-11-01-preview' = { + name: name + location: location + tags: tags + sku: sku + properties: { + adminUserEnabled: adminUserEnabled + anonymousPullEnabled: anonymousPullEnabled + dataEndpointEnabled: dataEndpointEnabled + encryption: encryption + metadataSearch: metadataSearch + networkRuleBypassOptions: networkRuleBypassOptions + policies:{ + quarantinePolicy: quarantinePolicy + trustPolicy: trustPolicy + retentionPolicy: retentionPolicy + exportPolicy: exportPolicy + azureADAuthenticationAsArmPolicy: azureADAuthenticationAsArmPolicy + softDeletePolicy: softDeletePolicy + } + publicNetworkAccess: publicNetworkAccess + zoneRedundancy: zoneRedundancy + } + + resource scopeMap 'scopeMaps' = [for scopeMap in scopeMaps: { + name: scopeMap.name + properties: scopeMap.properties + }] +} + +resource diagnostics 'Microsoft.Insights/diagnosticSettings@2021-05-01-preview' = if (!empty(workspaceId)) { + name: 'registry-diagnostics' + scope: containerRegistry + properties: { + workspaceId: workspaceId + logs: [ + { + category: 'ContainerRegistryRepositoryEvents' + enabled: true + } + { + category: 'ContainerRegistryLoginEvents' + enabled: true + } + ] + metrics: [ + { + category: 'AllMetrics' + enabled: true + timeGrain: 'PT1M' + } + ] + } +} + +output id string = containerRegistry.id +output loginServer string = containerRegistry.properties.loginServer +output name string = containerRegistry.name diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/networking/vnet.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/networking/vnet.bicep new file mode 100644 index 0000000..57b5051 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/networking/vnet.bicep @@ -0,0 +1,50 @@ +@description('The name of the Virtual Network') +param name string + +@description('The Azure region where the Virtual Network should exist') +param location string = resourceGroup().location + +@description('Optional tags for the resources') +param tags object = {} + +@description('The address prefixes of the Virtual Network') +param addressPrefixes array = ['10.0.0.0/16'] + +@description('The subnets to create in the Virtual Network') +param subnets array = [ + { + name: 'infrastructure-subnet' + properties: { + addressPrefix: '10.0.0.0/21' + delegations: [] + privateEndpointNetworkPolicies: 'Disabled' + privateLinkServiceNetworkPolicies: 'Enabled' + } + } + { + name: 'workload-subnet' + properties: { + addressPrefix: '10.0.8.0/21' + delegations: [] + privateEndpointNetworkPolicies: 'Disabled' + privateLinkServiceNetworkPolicies: 'Enabled' + } + } +] + +resource vnet 'Microsoft.Network/virtualNetworks@2022-07-01' = { + name: name + location: location + tags: tags + properties: { + addressSpace: { + addressPrefixes: addressPrefixes + } + subnets: subnets + } +} + +output id string = vnet.id +output name string = vnet.name +output infrastructureSubnetId string = resourceId('Microsoft.Network/virtualNetworks/subnets', name, 'infrastructure-subnet') +output workloadSubnetId string = resourceId('Microsoft.Network/virtualNetworks/subnets', name, 'workload-subnet') diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/security/registry-access.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/security/registry-access.bicep new file mode 100644 index 0000000..fc66837 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/security/registry-access.bicep @@ -0,0 +1,19 @@ +metadata description = 'Assigns ACR Pull permissions to access an Azure Container Registry.' +param containerRegistryName string +param principalId string + +var acrPullRole = subscriptionResourceId('Microsoft.Authorization/roleDefinitions', '7f951dda-4ed3-4680-a7ca-43fe172d538d') + +resource aksAcrPull 'Microsoft.Authorization/roleAssignments@2022-04-01' = { + scope: containerRegistry // Use when specifying a scope that is different than the deployment scope + name: guid(subscription().id, resourceGroup().id, principalId, acrPullRole) + properties: { + roleDefinitionId: acrPullRole + principalType: 'ServicePrincipal' + principalId: principalId + } +} + +resource containerRegistry 'Microsoft.ContainerRegistry/registries@2023-01-01-preview' existing = { + name: containerRegistryName +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/security/role.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/security/role.bicep new file mode 100644 index 0000000..0b30cfd --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/core/security/role.bicep @@ -0,0 +1,21 @@ +metadata description = 'Creates a role assignment for a service principal.' +param principalId string + +@allowed([ + 'Device' + 'ForeignGroup' + 'Group' + 'ServicePrincipal' + 'User' +]) +param principalType string = 'ServicePrincipal' +param roleDefinitionId string + +resource role 'Microsoft.Authorization/roleAssignments@2022-04-01' = { + name: guid(subscription().id, resourceGroup().id, principalId, roleDefinitionId) + properties: { + principalId: principalId + principalType: principalType + roleDefinitionId: resourceId('Microsoft.Authorization/roleDefinitions', roleDefinitionId) + } +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/main.bicep b/samples/scenarios/WorkItemFilteringSplitActivities/infra/main.bicep new file mode 100644 index 0000000..5bcab16 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/main.bicep @@ -0,0 +1,209 @@ +targetScope = 'subscription' + +@minLength(1) +@maxLength(64) +@description('Name of the the environment which is used to generate a short unique hash used in all resources.') +param environmentName string + +@minLength(1) +@description('Primary location for all resources') +param location string + +@description('Id of the user or app to assign application roles') +param principalId string = '' + +param containerAppsEnvName string = '' +param containerAppsAppName string = '' +param containerRegistryName string = '' +param dtsLocation string = 'centralus' +param dtsSkuName string = 'Consumption' +param dtsCapacity int = 1 +param dtsName string = '' +param taskHubName string = '' + +param clientServiceName string = 'client' +param orchestratorWorkerServiceName string = 'orchestrator-worker' +param validatorWorkerServiceName string = 'validator-worker' +param shipperWorkerServiceName string = 'shipper-worker' + +param resourceGroupName string = '' + +var abbrs = loadJsonContent('./abbreviations.json') + +var tags = { + 'azd-env-name': environmentName +} + +#disable-next-line no-unused-vars +var resourceToken = toLower(uniqueString(subscription().id, environmentName, location)) + +// Resource Group +resource rg 'Microsoft.Resources/resourceGroups@2021-04-01' = { + name: !empty(resourceGroupName) ? resourceGroupName : '${abbrs.resourcesResourceGroups}${environmentName}' + location: location + tags: tags +} + +// User-assigned managed identity for all container apps +module identity './app/user-assigned-identity.bicep' = { + name: 'identity' + scope: rg + params: { + name: 'dts-ca-identity' + } +} + +// Assign DTS Worker/Client role to the managed identity +module identityAssignDTS './core/security/role.bicep' = { + name: 'identityAssignDTS' + scope: rg + params: { + principalId: identity.outputs.principalId + roleDefinitionId: '0ad04412-c4d5-4796-b79c-f76d14c8d402' + principalType: 'ServicePrincipal' + } +} + +// Assign DTS role to the deploying user (for dashboard access) +module identityAssignDTSDash './core/security/role.bicep' = { + name: 'identityAssignDTSDash' + scope: rg + params: { + principalId: principalId + roleDefinitionId: '0ad04412-c4d5-4796-b79c-f76d14c8d402' + principalType: 'User' + } +} + +// Virtual network +module vnet './core/networking/vnet.bicep' = { + name: 'vnet' + scope: rg + params: { + name: '${abbrs.networkVirtualNetworks}${resourceToken}' + location: location + tags: tags + } +} + +// Container Apps Environment + Container Registry +module containerAppsEnv './core/host/container-apps.bicep' = { + name: 'container-apps' + scope: rg + params: { + name: 'app' + containerAppsEnvironmentName: !empty(containerAppsEnvName) ? containerAppsEnvName : '${abbrs.appManagedEnvironments}${resourceToken}' + containerRegistryName: !empty(containerRegistryName) ? containerRegistryName : '${abbrs.containerRegistryRegistries}${resourceToken}' + location: location + subnetResourceId: vnet.outputs.infrastructureSubnetId + loadBalancerType: 'External' + } +} + +// Durable Task Scheduler + Task Hub +module dts './app/dts.bicep' = { + scope: rg + name: 'dtsResource' + params: { + name: !empty(dtsName) ? dtsName : '${abbrs.dts}${resourceToken}' + taskhubname: !empty(taskHubName) ? taskHubName : '${abbrs.taskhub}${resourceToken}' + location: dtsLocation + tags: tags + ipAllowlist: ['0.0.0.0/0'] + skuName: dtsSkuName + skuCapacity: dtsCapacity + } +} + +// Client — schedules orchestrations and polls for results +module client 'app/app.bicep' = { + name: clientServiceName + scope: rg + params: { + appName: !empty(containerAppsAppName) ? '${containerAppsAppName}-client' : '${abbrs.appContainerApps}${resourceToken}-client' + containerAppsEnvironmentName: containerAppsEnv.outputs.environmentName + containerRegistryName: containerAppsEnv.outputs.registryName + userAssignedManagedIdentity: { + resourceId: identity.outputs.resourceId + clientId: identity.outputs.clientId + } + location: location + tags: tags + serviceName: 'client' + identityName: identity.outputs.name + dtsEndpoint: dts.outputs.dts_URL + taskHubName: dts.outputs.TASKHUB_NAME + } +} + +// Orchestrator Worker — handles orchestrations only +module orchestratorWorker 'app/app.bicep' = { + name: orchestratorWorkerServiceName + scope: rg + params: { + appName: !empty(containerAppsAppName) ? '${containerAppsAppName}-orchestrator' : '${abbrs.appContainerApps}${resourceToken}-orchestrator' + containerAppsEnvironmentName: containerAppsEnv.outputs.environmentName + containerRegistryName: containerAppsEnv.outputs.registryName + userAssignedManagedIdentity: { + resourceId: identity.outputs.resourceId + clientId: identity.outputs.clientId + } + location: location + tags: tags + serviceName: 'orchestrator-worker' + identityName: identity.outputs.name + dtsEndpoint: dts.outputs.dts_URL + taskHubName: dts.outputs.TASKHUB_NAME + workItemType: 'Orchestration' + } +} + +// Validator Worker — handles ValidateOrder activity only +module validatorWorker 'app/app.bicep' = { + name: validatorWorkerServiceName + scope: rg + params: { + appName: !empty(containerAppsAppName) ? '${containerAppsAppName}-validator' : '${abbrs.appContainerApps}${resourceToken}-validator' + containerAppsEnvironmentName: containerAppsEnv.outputs.environmentName + containerRegistryName: containerAppsEnv.outputs.registryName + userAssignedManagedIdentity: { + resourceId: identity.outputs.resourceId + clientId: identity.outputs.clientId + } + location: location + tags: tags + serviceName: 'validator-worker' + identityName: identity.outputs.name + dtsEndpoint: dts.outputs.dts_URL + taskHubName: dts.outputs.TASKHUB_NAME + workItemType: 'Activity' + } +} + +// Shipper Worker — handles ShipOrder activity only +module shipperWorker 'app/app.bicep' = { + name: shipperWorkerServiceName + scope: rg + params: { + appName: !empty(containerAppsAppName) ? '${containerAppsAppName}-shipper' : '${abbrs.appContainerApps}${resourceToken}-shipper' + containerAppsEnvironmentName: containerAppsEnv.outputs.environmentName + containerRegistryName: containerAppsEnv.outputs.registryName + userAssignedManagedIdentity: { + resourceId: identity.outputs.resourceId + clientId: identity.outputs.clientId + } + location: location + tags: tags + serviceName: 'shipper-worker' + identityName: identity.outputs.name + dtsEndpoint: dts.outputs.dts_URL + taskHubName: dts.outputs.TASKHUB_NAME + workItemType: 'Activity' + } +} + +output AZURE_LOCATION string = location +output AZURE_TENANT_ID string = tenant().tenantId +output AZURE_CONTAINER_REGISTRY_ENDPOINT string = containerAppsEnv.outputs.registryLoginServer +output AZURE_CONTAINER_REGISTRY_NAME string = containerAppsEnv.outputs.registryName +output AZURE_USER_ASSIGNED_IDENTITY_NAME string = identity.outputs.name diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/infra/main.parameters.json b/samples/scenarios/WorkItemFilteringSplitActivities/infra/main.parameters.json new file mode 100644 index 0000000..c8d3453 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/infra/main.parameters.json @@ -0,0 +1,15 @@ +{ + "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentParameters.json#", + "contentVersion": "1.0.0.0", + "parameters": { + "environmentName": { + "value": "${AZURE_ENV_NAME}" + }, + "location": { + "value": "${AZURE_LOCATION}" + }, + "principalId": { + "value": "${AZURE_PRINCIPAL_ID}" + } + } +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Client.csproj b/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Client.csproj new file mode 100644 index 0000000..0a6f1a8 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Client.csproj @@ -0,0 +1,19 @@ + + + + Exe + net10.0 + enable + enable + e72646d1-e2e4-47d5-930c-20477d4a28bf + + + + + + + + + + + diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Dockerfile b/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Dockerfile new file mode 100644 index 0000000..b1c1bf2 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Dockerfile @@ -0,0 +1,14 @@ +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build +WORKDIR /src + +COPY ["Client.csproj", "./"] +RUN dotnet restore + +COPY . . +RUN dotnet publish -c Release -o /app/publish + +FROM mcr.microsoft.com/dotnet/runtime:10.0 AS final +WORKDIR /app +COPY --from=build /app/publish . + +ENTRYPOINT ["dotnet", "Client.dll"] diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Program.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Program.cs new file mode 100644 index 0000000..18f02be --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/Client/Program.cs @@ -0,0 +1,139 @@ +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +// Configure logging +using ILoggerFactory loggerFactory = LoggerFactory.Create(builder => +{ + builder.AddSimpleConsole(options => + { + options.SingleLine = true; + options.TimestampFormat = "HH:mm:ss "; + }); + builder.SetMinimumLevel(LogLevel.Information); +}); + +ILogger logger = loggerFactory.CreateLogger("Client"); + +logger.LogInformation("=== Work Item Filtering Demo — Client ==="); + +// Build connection string +string endpoint = Environment.GetEnvironmentVariable("ENDPOINT") ?? "http://localhost:8080"; +string taskHubName = Environment.GetEnvironmentVariable("TASKHUB") ?? "default"; +string? managedIdentityClientId = Environment.GetEnvironmentVariable("AZURE_MANAGED_IDENTITY_CLIENT_ID"); +bool isLocalEmulator = endpoint == "http://localhost:8080"; + +string connectionString = isLocalEmulator + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=None" + : !string.IsNullOrEmpty(managedIdentityClientId) + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=ManagedIdentity;ClientID={managedIdentityClientId}" + : $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=DefaultAzure"; + +logger.LogInformation("Connection: {ConnectionString}", connectionString); + +// Create the Durable Task client +ServiceCollection services = new(); +services.AddLogging(lb => +{ + lb.AddSimpleConsole(o => { o.SingleLine = true; o.TimestampFormat = "HH:mm:ss "; }); + lb.SetMinimumLevel(LogLevel.Information); +}); +services.AddDurableTaskClient(options => +{ + options.UseDurableTaskScheduler(connectionString); +}); + +await using ServiceProvider serviceProvider = services.BuildServiceProvider(); +DurableTaskClient client = serviceProvider.GetRequiredService(); + +// Run in a loop: schedule a batch of orchestrations every 30 seconds for 15 minutes +const int orchestrationsPerBatch = 3; +TimeSpan interval = TimeSpan.FromSeconds(30); +TimeSpan totalDuration = TimeSpan.FromMinutes(10); +DateTime deadline = DateTime.UtcNow + totalDuration; + +int totalCompleted = 0; +int totalFailed = 0; +int batchNumber = 0; + +logger.LogInformation("Will schedule {Count} orchestrations every {Interval}s for {Duration} minutes.", + orchestrationsPerBatch, interval.TotalSeconds, totalDuration.TotalMinutes); +logger.LogInformation("(Make sure the Orchestrator, Validator, and Shipper workers are all running)\n"); + +while (DateTime.UtcNow < deadline) +{ + batchNumber++; + logger.LogInformation("--- Batch #{Batch} at {Time:HH:mm:ss} ---", batchNumber, DateTime.UtcNow); + + var instanceIds = new List(); + for (int i = 1; i <= orchestrationsPerBatch; i++) + { + string orderId = $"ORD-B{batchNumber:D3}-{i:D3}"; + logger.LogInformation("Scheduling orchestration with orderId='{OrderId}'...", orderId); + + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync( + "OrderProcessingOrchestration", + orderId); + + instanceIds.Add(instanceId); + logger.LogInformation(" -> Scheduled with InstanceId={InstanceId}", instanceId); + } + + // Wait for all orchestrations in this batch to complete + int batchCompleted = 0; + int batchFailed = 0; + + foreach (string id in instanceIds) + { + try + { + OrchestrationMetadata result = await client.WaitForInstanceCompletionAsync( + id, getInputsAndOutputs: true, CancellationToken.None); + + if (result.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + batchCompleted++; + logger.LogInformation( + "COMPLETED | InstanceId={InstanceId} | Output: {Output}", + result.InstanceId, result.ReadOutputAs()); + } + else + { + batchFailed++; + logger.LogError( + "FAILED | InstanceId={InstanceId} | Status={Status} | Error: {Error}", + result.InstanceId, result.RuntimeStatus, result.FailureDetails?.ErrorMessage); + } + } + catch (Exception ex) + { + batchFailed++; + logger.LogError(ex, "Error waiting for orchestration {Id}", id); + } + } + + totalCompleted += batchCompleted; + totalFailed += batchFailed; + + logger.LogInformation("Batch #{Batch} results: {Completed} completed, {Failed} failed", + batchNumber, batchCompleted, batchFailed); + + // Wait for the next interval (unless we've passed the deadline) + if (DateTime.UtcNow < deadline) + { + TimeSpan remaining = deadline - DateTime.UtcNow; + TimeSpan waitTime = remaining < interval ? remaining : interval; + logger.LogInformation("Next batch in {Seconds:F0}s (deadline in {Remaining:F1} min)\n", + waitTime.TotalSeconds, remaining.TotalMinutes); + await Task.Delay(waitTime); + } +} + +logger.LogInformation("\n=== FINAL RESULTS: {Completed} completed, {Failed} failed across {Batches} batches ===", + totalCompleted, totalFailed, batchNumber); + +// Keep the process alive so Container Apps doesn't mark it as failed +logger.LogInformation("Demo complete. Staying alive — press Ctrl+C to exit."); +await Task.Delay(Timeout.Infinite, CancellationToken.None); diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/Dockerfile b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/Dockerfile new file mode 100644 index 0000000..b0c232b --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/Dockerfile @@ -0,0 +1,14 @@ +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build +WORKDIR /src + +COPY ["OrchestratorWorker.csproj", "./"] +RUN dotnet restore + +COPY . . +RUN dotnet publish -c Release -o /app/publish + +FROM mcr.microsoft.com/dotnet/runtime:10.0 AS final +WORKDIR /app +COPY --from=build /app/publish . + +ENTRYPOINT ["dotnet", "OrchestratorWorker.dll"] diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/OrchestratorWorker.csproj b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/OrchestratorWorker.csproj new file mode 100644 index 0000000..e229c4f --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/OrchestratorWorker.csproj @@ -0,0 +1,20 @@ + + + + Exe + net10.0 + enable + enable + 0e9c5fe0-ec2a-480d-ace3-a3266e1d14b4 + + + + + + + + + + + + diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/OrderProcessingOrchestration.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/OrderProcessingOrchestration.cs new file mode 100644 index 0000000..de5cbe0 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/OrderProcessingOrchestration.cs @@ -0,0 +1,45 @@ +using Microsoft.DurableTask; +using Microsoft.Extensions.Logging; + +namespace OrchestratorWorker; + +/// +/// A simple orchestration that calls two activities sequentially: +/// 1. ValidateOrder (handled by Worker A) +/// 2. ShipOrder (handled by Worker B) +/// +/// Because each activity is registered in a different worker process, DTS routes +/// each activity work item to the correct worker via work item filtering. +/// +[DurableTask(nameof(OrderProcessingOrchestration))] +public class OrderProcessingOrchestration : TaskOrchestrator +{ + public override async Task RunAsync(TaskOrchestrationContext context, string orderId) + { + ILogger logger = context.CreateReplaySafeLogger(); + + logger.LogInformation( + "[Orchestrator] Orchestration | Name=OrderProcessingOrchestration | InstanceId={InstanceId} | Processing order '{OrderId}'", + context.InstanceId, orderId); + + // Step 1: Validate the order (routed to Validator Worker) + logger.LogInformation( + "[Orchestrator] Orchestration | InstanceId={InstanceId} | Dispatching ValidateOrder to Validator Worker...", + context.InstanceId); + string validationResult = await context.CallActivityAsync("ValidateOrder", orderId); + + // Step 2: Ship the order (routed to Shipper Worker) + logger.LogInformation( + "[Orchestrator] Orchestration | InstanceId={InstanceId} | Dispatching ShipOrder to Shipper Worker...", + context.InstanceId); + string shippingResult = await context.CallActivityAsync("ShipOrder", orderId); + + string combined = $"Order '{orderId}' => Validation: [{validationResult}], Shipping: [{shippingResult}]"; + + logger.LogInformation( + "[Orchestrator] Orchestration | InstanceId={InstanceId} | Completed: {Result}", + context.InstanceId, combined); + + return combined; + } +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/Program.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/Program.cs new file mode 100644 index 0000000..d4b587a --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/OrchestratorWorker/Program.cs @@ -0,0 +1,56 @@ +using Microsoft.DurableTask; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +const string WorkerName = "Orchestrator Worker"; + +HostApplicationBuilder builder = Host.CreateApplicationBuilder(); + +builder.Logging.AddSimpleConsole(options => +{ + options.SingleLine = true; + options.TimestampFormat = "HH:mm:ss "; +}); +builder.Logging.SetMinimumLevel(LogLevel.Information); + +// Build a logger for startup +using ILoggerFactory startupLoggerFactory = LoggerFactory.Create(lb => +{ + lb.AddSimpleConsole(o => { o.SingleLine = true; o.TimestampFormat = "HH:mm:ss "; }); + lb.SetMinimumLevel(LogLevel.Information); +}); +ILogger logger = startupLoggerFactory.CreateLogger(WorkerName); + +string endpoint = Environment.GetEnvironmentVariable("ENDPOINT") ?? "http://localhost:8080"; +string taskHubName = Environment.GetEnvironmentVariable("TASKHUB") ?? "default"; +string? managedIdentityClientId = Environment.GetEnvironmentVariable("AZURE_MANAGED_IDENTITY_CLIENT_ID"); +bool isLocalEmulator = endpoint == "http://localhost:8080"; + +string connectionString = isLocalEmulator + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=None" + : !string.IsNullOrEmpty(managedIdentityClientId) + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=ManagedIdentity;ClientID={managedIdentityClientId}" + : $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=DefaultAzure"; + +logger.LogInformation("[{Worker}] Connection: {ConnectionString}", WorkerName, connectionString); +logger.LogInformation("[{Worker}] This worker registers ONLY the orchestration. No activities.", WorkerName); + +builder.Services.AddDurableTaskWorker() + .AddTasks(registry => + { + // Only the orchestration is registered here. + // Work item filters are auto-generated from the registry, so this worker + // will ONLY receive orchestration work items — never activity work items. + registry.AddAllGeneratedTasks(); + }) + .UseDurableTaskScheduler(connectionString); + +IHost host = builder.Build(); +logger = host.Services.GetRequiredService>(); + +logger.LogInformation("[{Worker}] Starting... waiting for orchestration work items only.", WorkerName); + +await host.RunAsync(); diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/Dockerfile b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/Dockerfile new file mode 100644 index 0000000..f93a630 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/Dockerfile @@ -0,0 +1,14 @@ +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build +WORKDIR /src + +COPY ["ShipperWorker.csproj", "./"] +RUN dotnet restore + +COPY . . +RUN dotnet publish -c Release -o /app/publish + +FROM mcr.microsoft.com/dotnet/runtime:10.0 AS final +WORKDIR /app +COPY --from=build /app/publish . + +ENTRYPOINT ["dotnet", "ShipperWorker.dll"] diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/Program.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/Program.cs new file mode 100644 index 0000000..07c3a18 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/Program.cs @@ -0,0 +1,55 @@ +using Microsoft.DurableTask; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +const string WorkerName = "Shipper Worker"; + +HostApplicationBuilder builder = Host.CreateApplicationBuilder(); + +builder.Logging.AddSimpleConsole(options => +{ + options.SingleLine = true; + options.TimestampFormat = "HH:mm:ss "; +}); +builder.Logging.SetMinimumLevel(LogLevel.Information); + +using ILoggerFactory startupLoggerFactory = LoggerFactory.Create(lb => +{ + lb.AddSimpleConsole(o => { o.SingleLine = true; o.TimestampFormat = "HH:mm:ss "; }); + lb.SetMinimumLevel(LogLevel.Information); +}); +ILogger logger = startupLoggerFactory.CreateLogger(WorkerName); + +string endpoint = Environment.GetEnvironmentVariable("ENDPOINT") ?? "http://localhost:8080"; +string taskHubName = Environment.GetEnvironmentVariable("TASKHUB") ?? "default"; +string? managedIdentityClientId = Environment.GetEnvironmentVariable("AZURE_MANAGED_IDENTITY_CLIENT_ID"); +bool isLocalEmulator = endpoint == "http://localhost:8080"; + +string connectionString = isLocalEmulator + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=None" + : !string.IsNullOrEmpty(managedIdentityClientId) + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=ManagedIdentity;ClientID={managedIdentityClientId}" + : $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=DefaultAzure"; + +logger.LogInformation("[{Worker}] Connection: {ConnectionString}", WorkerName, connectionString); +logger.LogInformation("[{Worker}] This worker registers ONLY the ShipOrder activity.", WorkerName); + +builder.Services.AddDurableTaskWorker() + .AddTasks(registry => + { + // Only ShipOrder is registered here. + // Work item filters are auto-generated from the registry, so this worker + // will ONLY receive ShipOrder activity work items. + registry.AddAllGeneratedTasks(); + }) + .UseDurableTaskScheduler(connectionString); + +IHost host = builder.Build(); +logger = host.Services.GetRequiredService>(); + +logger.LogInformation("[{Worker}] Starting... waiting for ShipOrder activity work items only.", WorkerName); + +await host.RunAsync(); diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/ShipOrder.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/ShipOrder.cs new file mode 100644 index 0000000..2dafebf --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/ShipOrder.cs @@ -0,0 +1,36 @@ +using Microsoft.DurableTask; +using Microsoft.Extensions.Logging; + +namespace ShipperWorker; + +/// +/// Ships an order. This activity is registered only in Worker B, +/// so DTS will route ShipOrder work items exclusively to Worker B. +/// +[DurableTask(nameof(ShipOrder))] +public class ShipOrder : TaskActivity +{ + readonly ILogger logger; + + public ShipOrder(ILoggerFactory loggerFactory) + { + this.logger = loggerFactory.CreateLogger(); + } + + public override Task RunAsync(TaskActivityContext context, string orderId) + { + this.logger.LogInformation( + "[Shipper] Activity | Name=ShipOrder | InstanceId={InstanceId} | Shipping order '{OrderId}'...", + context.InstanceId, orderId); + + // Simulate shipping + string trackingNumber = $"TRACK-{orderId}-{Random.Shared.Next(1000, 9999)}"; + string result = $"Shipped with tracking {trackingNumber}"; + + this.logger.LogInformation( + "[Shipper] Activity | Name=ShipOrder | InstanceId={InstanceId} | Result: {Result}", + context.InstanceId, result); + + return Task.FromResult(result); + } +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/ShipperWorker.csproj b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/ShipperWorker.csproj new file mode 100644 index 0000000..0051ede --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ShipperWorker/ShipperWorker.csproj @@ -0,0 +1,20 @@ + + + + Exe + net10.0 + enable + enable + 2200f67e-b6f7-4c86-b2f0-745ddf026a3c + + + + + + + + + + + + diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/Dockerfile b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/Dockerfile new file mode 100644 index 0000000..9dcdac6 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/Dockerfile @@ -0,0 +1,14 @@ +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build +WORKDIR /src + +COPY ["ValidatorWorker.csproj", "./"] +RUN dotnet restore + +COPY . . +RUN dotnet publish -c Release -o /app/publish + +FROM mcr.microsoft.com/dotnet/runtime:10.0 AS final +WORKDIR /app +COPY --from=build /app/publish . + +ENTRYPOINT ["dotnet", "ValidatorWorker.dll"] diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/Program.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/Program.cs new file mode 100644 index 0000000..f2d411a --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/Program.cs @@ -0,0 +1,55 @@ +using Microsoft.DurableTask; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +const string WorkerName = "Validator Worker"; + +HostApplicationBuilder builder = Host.CreateApplicationBuilder(); + +builder.Logging.AddSimpleConsole(options => +{ + options.SingleLine = true; + options.TimestampFormat = "HH:mm:ss "; +}); +builder.Logging.SetMinimumLevel(LogLevel.Information); + +using ILoggerFactory startupLoggerFactory = LoggerFactory.Create(lb => +{ + lb.AddSimpleConsole(o => { o.SingleLine = true; o.TimestampFormat = "HH:mm:ss "; }); + lb.SetMinimumLevel(LogLevel.Information); +}); +ILogger logger = startupLoggerFactory.CreateLogger(WorkerName); + +string endpoint = Environment.GetEnvironmentVariable("ENDPOINT") ?? "http://localhost:8080"; +string taskHubName = Environment.GetEnvironmentVariable("TASKHUB") ?? "default"; +string? managedIdentityClientId = Environment.GetEnvironmentVariable("AZURE_MANAGED_IDENTITY_CLIENT_ID"); +bool isLocalEmulator = endpoint == "http://localhost:8080"; + +string connectionString = isLocalEmulator + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=None" + : !string.IsNullOrEmpty(managedIdentityClientId) + ? $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=ManagedIdentity;ClientID={managedIdentityClientId}" + : $"Endpoint={endpoint};TaskHub={taskHubName};Authentication=DefaultAzure"; + +logger.LogInformation("[{Worker}] Connection: {ConnectionString}", WorkerName, connectionString); +logger.LogInformation("[{Worker}] This worker registers ONLY the ValidateOrder activity.", WorkerName); + +builder.Services.AddDurableTaskWorker() + .AddTasks(registry => + { + // Only ValidateOrder is registered here. + // Work item filters are auto-generated from the registry, so this worker + // will ONLY receive ValidateOrder activity work items. + registry.AddAllGeneratedTasks(); + }) + .UseDurableTaskScheduler(connectionString); + +IHost host = builder.Build(); +logger = host.Services.GetRequiredService>(); + +logger.LogInformation("[{Worker}] Starting... waiting for ValidateOrder activity work items only.", WorkerName); + +await host.RunAsync(); diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/ValidateOrder.cs b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/ValidateOrder.cs new file mode 100644 index 0000000..89bbdd1 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/ValidateOrder.cs @@ -0,0 +1,35 @@ +using Microsoft.DurableTask; +using Microsoft.Extensions.Logging; + +namespace ValidatorWorker; + +/// +/// Validates an incoming order. This activity is registered only in Worker A, +/// so DTS will route ValidateOrder work items exclusively to Worker A. +/// +[DurableTask(nameof(ValidateOrder))] +public class ValidateOrder : TaskActivity +{ + readonly ILogger logger; + + public ValidateOrder(ILoggerFactory loggerFactory) + { + this.logger = loggerFactory.CreateLogger(); + } + + public override Task RunAsync(TaskActivityContext context, string orderId) + { + this.logger.LogInformation( + "[Validator] Activity | Name=ValidateOrder | InstanceId={InstanceId} | Validating order '{OrderId}'...", + context.InstanceId, orderId); + + // Simulate validation + string result = $"Order {orderId} is valid"; + + this.logger.LogInformation( + "[Validator] Activity | Name=ValidateOrder | InstanceId={InstanceId} | Result: {Result}", + context.InstanceId, result); + + return Task.FromResult(result); + } +} diff --git a/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/ValidatorWorker.csproj b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/ValidatorWorker.csproj new file mode 100644 index 0000000..68ab504 --- /dev/null +++ b/samples/scenarios/WorkItemFilteringSplitActivities/src/ValidatorWorker/ValidatorWorker.csproj @@ -0,0 +1,20 @@ + + + + Exe + net10.0 + enable + enable + 3719e2c5-9a8b-4350-91ca-67fcc5c05835 + + + + + + + + + + + +