Skip to content

Commit

Permalink
Merge branch 'v1.15' into ephemeral-nonha
Browse files Browse the repository at this point in the history
  • Loading branch information
elena-kolevska authored Dec 3, 2024
2 parents 7a46161 + 56d4a41 commit 0930d26
Show file tree
Hide file tree
Showing 19 changed files with 202 additions and 23 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -336,14 +336,13 @@ Status | Description
`RETRY` | Message to be retried by Dapr
`DROP` | Warning is logged and message is dropped

Please refer [Expected HTTP Response for Bulk Subscribe]({{< ref pubsub_api.md >}}) for further insights on response.
Refer to [Expected HTTP Response for Bulk Subscribe]({{< ref pubsub_api.md >}}) for further insights on response.

### Example

Please refer following code samples for how to use Bulk Subscribe:

{{< tabs "Java" "JavaScript" ".NET" >}}
The following code examples demonstrate how to use Bulk Subscribe.

{{< tabs "Java" "JavaScript" ".NET" "Python" >}}
{{% codetab %}}

```java
Expand Down Expand Up @@ -471,7 +470,50 @@ public class BulkMessageController : ControllerBase

{{% /codetab %}}

{{% codetab %}}
Currently, you can only bulk subscribe in Python using an HTTP client.

```python
import json
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
# Define the bulk subscribe configuration
subscriptions = [{
"pubsubname": "pubsub",
"topic": "TOPIC_A",
"route": "/checkout",
"bulkSubscribe": {
"enabled": True,
"maxMessagesCount": 3,
"maxAwaitDurationMs": 40
}
}]
print('Dapr pub/sub is subscribed to: ' + json.dumps(subscriptions))
return jsonify(subscriptions)
# Define the endpoint to handle incoming messages
@app.route('/checkout', methods=['POST'])
def checkout():
messages = request.json
print(messages)
for message in messages:
print(f"Received message: {message}")
return json.dumps({'success': True}), 200, {'ContentType': 'application/json'}
if __name__ == '__main__':
app.run(port=5000)
```

{{% /codetab %}}

{{< /tabs >}}

## How components handle publishing and subscribing to bulk messages

For event publish/subscribe, two kinds of network transfers are involved.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ Currently, you can experience this actors quickstart using the .NET SDK.
As a quick overview of the .NET actors quickstart:

1. Using a `SmartDevice.Service` microservice, you host:
- Two `SmartDectectorActor` smoke alarm objects
- Two `SmokeDetectorActor` smoke alarm objects
- A `ControllerActor` object that commands and controls the smart devices
1. Using a `SmartDevice.Client` console app, the client app interacts with each actor, or the controller, to perform actions in aggregate.
1. The `SmartDevice.Interfaces` contains the shared interfaces and data types used by both the service and client apps.
Expand Down Expand Up @@ -119,7 +119,7 @@ If you have Zipkin configured for Dapr locally on your machine, you can view the

When you ran the client app, a few things happened:

1. Two `SmartDetectorActor` actors were [created in the client application](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/client/Program.cs) and initialized with object state with:
1. Two `SmokeDetectorActor` actors were [created in the client application](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/client/Program.cs) and initialized with object state with:
- `ActorProxy.Create<ISmartDevice>(actorId, actorType)`
- `proxySmartDevice.SetDataAsync(data)`

Expand Down Expand Up @@ -177,7 +177,7 @@ When you ran the client app, a few things happened:
Console.WriteLine($"Device 2 state: {storedDeviceData2}");
```

1. The [`DetectSmokeAsync` method of `SmartDetectorActor 1` is called](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/SmokeDetectorActor.cs#L70).
1. The [`DetectSmokeAsync` method of `SmokeDetectorActor 1` is called](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/SmokeDetectorActor.cs#L70).

```csharp
public async Task DetectSmokeAsync()
Expand Down Expand Up @@ -216,7 +216,7 @@ When you ran the client app, a few things happened:
await proxySmartDevice1.DetectSmokeAsync();
```

1. The [`SoundAlarm` methods](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/SmokeDetectorActor.cs#L78) of `SmartDetectorActor 1` and `2` are called.
1. The [`SoundAlarm` methods](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/SmokeDetectorActor.cs#L78) of `SmokeDetectorActor 1` and `2` are called.
```csharp
storedDeviceData1 = await proxySmartDevice1.GetDataAsync();
Expand All @@ -234,9 +234,9 @@ When you ran the client app, a few things happened:

For full context of the sample, take a look at the following code:

- [`SmartDetectorActor.cs`](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/SmokeDetectorActor.cs): Implements the smart device actors
- [`SmokeDetectorActor.cs`](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/SmokeDetectorActor.cs): Implements the smart device actors
- [`ControllerActor.cs`](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/service/ControllerActor.cs): Implements the controller actor that manages all devices
- [`ISmartDevice`](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/interfaces/ISmartDevice.cs): The method definitions and shared data types for each `SmartDetectorActor`
- [`ISmartDevice`](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/interfaces/ISmartDevice.cs): The method definitions and shared data types for each `SmokeDetectorActor`
- [`IController`](https://github.com/dapr/quickstarts/blob/master/actors/csharp/sdk/interfaces/IController.cs): The method definitions and shared data types for the `ControllerActor`
{{% /codetab %}}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ This guide walks you through installing an Elastic Kubernetes Service (EKS) clus
- [AWS CLI](https://aws.amazon.com/cli/)
- [eksctl](https://eksctl.io/)
- [An existing VPC and subnets](https://docs.aws.amazon.com/eks/latest/userguide/network_reqs.html)
- [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/)

## Deploy an EKS cluster

Expand All @@ -25,20 +26,57 @@ This guide walks you through installing an Elastic Kubernetes Service (EKS) clus
aws configure
```

1. Create an EKS cluster. To use a specific version of Kubernetes, use `--version` (1.13.x or newer version required).
1. Create a new file called `cluster-config.yaml` and add the content below to it, replacing `[your_cluster_name]`, `[your_cluster_region]`, and `[your_k8s_version]` with the appropriate values:

```yaml
apiVersion: eksctl.io/v1alpha5
kind: ClusterConfig

metadata:
name: [your_cluster_name]
region: [your_cluster_region]
version: [your_k8s_version]
tags:
karpenter.sh/discovery: [your_cluster_name]

iam:
withOIDC: true

managedNodeGroups:
- name: mng-od-4vcpu-8gb
desiredCapacity: 2
minSize: 1
maxSize: 5
instanceType: c5.xlarge
privateNetworking: true

addons:
- name: vpc-cni
attachPolicyARNs:
- arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy
- name: coredns
version: latest
- name: kube-proxy
version: latest
- name: aws-ebs-csi-driver
wellKnownPolicies:
ebsCSIController: true
```
1. Create the cluster by running the following command:
```bash
eksctl create cluster --name [your_eks_cluster_name] --region [your_aws_region] --version [kubernetes_version] --vpc-private-subnets [subnet_list_seprated_by_comma] --without-nodegroup
eksctl create cluster -f cluster.yaml
```

Change the values for `vpc-private-subnets` to meet your requirements. You can also add additional IDs. You must specify at least two subnet IDs. If you'd rather specify public subnets, you can change `--vpc-private-subnets` to `--vpc-public-subnets`.

1. Verify kubectl context:

1. Verify the kubectl context:

```bash
kubectl config current-context
```

## Add Dapr requirements for sidecar access and default storage class:

1. Update the security group rule to allow the EKS cluster to communicate with the Dapr Sidecar by creating an inbound rule for port 4000.

```bash
Expand All @@ -49,11 +87,37 @@ This guide walks you through installing an Elastic Kubernetes Service (EKS) clus
--source-group [your_security_group]
```

2. Add a default storage class if you don't have one:

```bash
kubectl patch storageclass gp2 -p '{"metadata": {"annotations":{"storageclass.kubernetes.io/is-default-class":"true"}}}'
```

## Install Dapr

Install Dapr on your cluster by running:

```bash
dapr init -k
```

You should see the following response:

```bash
⌛ Making the jump to hyperspace...
ℹ️ Note: To install Dapr using Helm, see here: https://docs.dapr.io/getting-started/install-dapr-kubernetes/#install-with-helm-advanced

ℹ️ Container images will be pulled from Docker Hub
✅ Deploying the Dapr control plane with latest version to your cluster...
✅ Deploying the Dapr dashboard with latest version to your cluster...
✅ Success! Dapr has been installed to namespace dapr-system. To verify, run `dapr status -k' in your terminal. To get started, go here: https://docs.dapr.io/getting-started
```
## Troubleshooting
### Access permissions
If you face any access permissions, make sure you are using the same AWS profile that was used to create the cluster. If needed, update the kubectl configuration with the correct profile:
If you face any access permissions, make sure you are using the same AWS profile that was used to create the cluster. If needed, update the kubectl configuration with the correct profile. More information [here](https://repost.aws/knowledge-center/eks-api-server-unauthorized-error):
```bash
aws eks --region [your_aws_region] update-kubeconfig --name [your_eks_cluster_name] --profile [your_profile_name]
Expand Down
2 changes: 1 addition & 1 deletion daprdocs/content/en/operations/observability/_index.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ weight: 60
description: See and measure the message calls to components and between networked services
---

[The following overview video and demo](https://www.youtube.com/live/0y7ne6teHT4?si=3bmNSSyIEIVSF-Ej&t=9931) demonstrates how observability in Dapr works.
[The following overview video and demo](https://www.youtube.com/watch?v=0y7ne6teHT4&t=12652s) demonstrates how observability in Dapr works.

<iframe width="560" height="315" src="https://www.youtube.com/embed/0y7ne6teHT4?si=iURnLk57t2zN-7zP&amp;start=12653" title="YouTube video player" style="padding-bottom:25px;" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" allowfullscreen></iframe>

Expand Down
22 changes: 22 additions & 0 deletions daprdocs/content/en/operations/resiliency/policies.md
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,15 @@ The following retry options are configurable:
| `duration` | Determines the time interval between retries. Only applies to the `constant` policy.<br/>Valid values are of the form `200ms`, `15s`, `2m`, etc.<br/> Defaults to `5s`.|
| `maxInterval` | Determines the maximum interval between retries to which the `exponential` back-off policy can grow.<br/>Additional retries always occur after a duration of `maxInterval`. Defaults to `60s`. Valid values are of the form `5s`, `1m`, `1m30s`, etc |
| `maxRetries` | The maximum number of retries to attempt. <br/>`-1` denotes an unlimited number of retries, while `0` means the request will not be retried (essentially behaving as if the retry policy were not set).<br/>Defaults to `-1`. |
| `matching.httpStatusCodes` | Optional: a comma-separated string of HTTP status codes or code ranges to retry. Status codes not listed are not retried.<br/>Valid values: 100-599, [Reference](https://developer.mozilla.org/en-US/docs/Web/HTTP/Status)<br/>Format: `<code>` or range `<start>-<end>`<br/>Example: "429,501-503"<br/>Default: empty string `""` or field is not set. Retries on all HTTP errors. |
| `matching.gRPCStatusCodes` | Optional: a comma-separated string of gRPC status codes or code ranges to retry. Status codes not listed are not retried.<br/>Valid values: 0-16, [Reference](https://grpc.io/docs/guides/status-codes/)<br/>Format: `<code>` or range `<start>-<end>`<br/>Example: "1,501-503"<br/>Default: empty string `""` or field is not set. Retries on all gRPC errors. |


{{% alert title="httpStatusCodes and gRPCStatusCodes format" color="warning" %}}
The field values should follow the format as specified in the field description or in the "Example 2" below.
An incorrectly formatted value will produce an error log ("Could not read resiliency policy") and `daprd` startup sequence will proceed.
{{% /alert %}}


The exponential back-off window uses the following formula:

Expand Down Expand Up @@ -77,7 +86,20 @@ spec:
maxRetries: -1 # Retry indefinitely
```

Example 2:

```yaml
spec:
policies:
retries:
retry5xxOnly:
policy: constant
duration: 5s
maxRetries: 3
matches:
httpStatusCodes: "429,500-599" # retry the HTTP status codes in this range. All others are not retried.
gRPCStatusCodes: "1-4,8-11,13,14" # retry gRPC status codes in these ranges and separate single codes.
```

## Circuit Breakers

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ After announcing a future breaking change, the change will happen in 2 releases
| Hazelcast PubSub Component | 1.9.0 | 1.11.0 |
| Twitter Binding Component | 1.10.0 | 1.11.0 |
| NATS Streaming PubSub Component | 1.11.0 | 1.13.0 |
| Workflows API Alpha1 `/v1.0-alpha1/workflows` being deprecated in favor of Workflow Client | 1.15.0 | 1.17.0 |

## Related links

Expand Down
2 changes: 1 addition & 1 deletion daprdocs/content/en/reference/api/pubsub_api.md
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,7 @@ other | warning is logged and all messages to be retried

## Message envelope

Dapr pub/sub adheres to version 1.0 of CloudEvents.
Dapr pub/sub adheres to [version 1.0 of CloudEvents](https://github.com/cloudevents/spec/blob/v1.0/spec.md).

## Related links

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,8 @@ spec:
value: "namespace"
- name: enableEntityManagement
value: "false"
- name: enableInOrderMessageDelivery
value: "false"
# The following four properties are needed only if enableEntityManagement is set to true
- name: resourceGroupName
value: "test-rg"
Expand Down Expand Up @@ -71,7 +73,8 @@ The above example uses secrets as plain strings. It is recommended to use a secr
| `eventHub` | Y* | Input/Output | The name of the Event Hubs hub ("topic"). Required if using Microsoft Entra ID authentication or if the connection string doesn't contain an `EntityPath` value | `mytopic` |
| `connectionString` | Y* | Input/Output | Connection string for the Event Hub or the Event Hub namespace.<br>* Mutally exclusive with `eventHubNamespace` field.<br>* Required when not using [Microsoft Entra ID Authentication]({{< ref "authenticating-azure.md" >}}) | `"Endpoint=sb://{EventHubNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={EventHub}"` or `"Endpoint=sb://{EventHubNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key}"`
| `eventHubNamespace` | Y* | Input/Output | The Event Hub Namespace name.<br>* Mutally exclusive with `connectionString` field.<br>* Required when using [Microsoft Entra ID Authentication]({{< ref "authenticating-azure.md" >}}) | `"namespace"`
| `enableEntityManagement` | N | Input/Output | Boolean value to allow management of the EventHub namespace and storage account. Default: `false` | `"true", "false"`
| `enableEntityManagement` | N | Input/Output | Boolean value to allow management of the EventHub namespace and storage account. Default: `false` | `"true"`, `"false"`
| `enableInOrderMessageDelivery` | N | Input/Output | Boolean value to allow messages to be delivered in the order in which they were posted. This assumes `partitionKey` is set when publishing or posting to ensure ordering across partitions. Default: `false` | `"true"`, `"false"`
| `resourceGroupName` | N | Input/Output | Name of the resource group the Event Hub namespace is part of. Required when entity management is enabled | `"test-rg"`
| `subscriptionID` | N | Input/Output | Azure subscription ID value. Required when entity management is enabled | `"azure subscription id"`
| `partitionCount` | N | Input/Output | Number of partitions for the new Event Hub namespace. Used only when entity management is enabled. Default: `"1"` | `"2"`
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -468,7 +468,7 @@ Apache Kafka supports the following bulk metadata options:

When invoking the Kafka pub/sub, its possible to provide an optional partition key by using the `metadata` query param in the request url.

The param name is `partitionKey`.
The param name can either be `partitionKey` or `__key`

Example:

Expand All @@ -484,7 +484,7 @@ curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.partiti

### Message headers

All other metadata key/value pairs (that are not `partitionKey`) are set as headers in the Kafka message. Here is an example setting a `correlationId` for the message.
All other metadata key/value pairs (that are not `partitionKey` or `__key`) are set as headers in the Kafka message. Here is an example setting a `correlationId` for the message.

```shell
curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.correlationId=myCorrelationID&metadata.partitionKey=key1 \
Expand All @@ -495,7 +495,51 @@ curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.correla
}
}'
```
### Kafka Pubsub special message headers received on consumer side

When consuming messages, special message metadata are being automatically passed as headers. These are:
- `__key`: the message key if available
- `__topic`: the topic for the message
- `__partition`: the partition number for the message
- `__offset`: the offset of the message in the partition
- `__timestamp`: the timestamp for the message

You can access them within the consumer endpoint as follows:
{{< tabs "Python (FastAPI)" >}}

{{% codetab %}}

```python
from fastapi import APIRouter, Body, Response, status
import json
import sys
app = FastAPI()
router = APIRouter()
@router.get('/dapr/subscribe')
def subscribe():
subscriptions = [{'pubsubname': 'pubsub',
'topic': 'my-topic',
'route': 'my_topic_subscriber',
}]
return subscriptions
@router.post('/my_topic_subscriber')
def my_topic_subscriber(
key: Annotated[str, Header(alias="__key")],
offset: Annotated[int, Header(alias="__offset")],
event_data=Body()):
print(f"key={key} - offset={offset} - data={event_data}", flush=True)
return Response(status_code=status.HTTP_200_OK)
app.include_router(router)
```

{{% /codetab %}}
## Receiving message headers with special characters

The consumer application may be required to receive message headers that include special characters, which may cause HTTP protocol validation errors.
Expand Down
Loading

0 comments on commit 0930d26

Please sign in to comment.