diff --git a/docs/cloud/connecting-services.mdx b/docs/cloud/connecting-services.mdx index 92c9fd98..f83dbb42 100644 --- a/docs/cloud/connecting-services.mdx +++ b/docs/cloud/connecting-services.mdx @@ -12,10 +12,10 @@ When using Restate Cloud, you can run your services anywhere: on Kubernetes, as The only requirement is that your services need to be reachable from Restate Cloud's infrastructure. You can connect your services to Restate Cloud in several ways, depending on where they run: -- Connect [Kubernetes services](/cloud/connecting-services#connecting-kubernetes-services) via a secure tunnel with Restate Operator. +- Connect [Kubernetes services](/cloud/connecting-services#kubernetes-services) via a secure tunnel with Restate Operator. - Connect [serverless functions (Vercel, Cloudflare Workers, Deno Deploy, etc.) or other public endpoints](/cloud/connecting-services#serverless-functions-and-other-public-endpoints), by signing requests with your cloud environment's public key. -- Connect [AWS Lambda functions](#connecting-aws-lambda-services), by granting Restate Cloud permission to assume a role in your AWS account. -- Connect [services in private environments](#connecting-services-in-private-environments), by setting up a tunnel. +- Connect [AWS Lambda functions](#aws-lambda-functions), by granting Restate Cloud permission to assume a role in your AWS account. +- Connect [services in private environments](#services-in-private-environments), by setting up a tunnel. If you prefer a video walkthrough, check out this webinar on getting started with cloud: diff --git a/docs/docs.json b/docs/docs.json index 3eca1a38..a23e1034 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -179,6 +179,7 @@ "services/deploy/kubernetes", "services/deploy/vercel", "services/deploy/lambda", + "services/deploy/cloud-run", "services/deploy/cloudflare-workers", "services/deploy/deno-deploy", "services/deploy/standalone" diff --git a/docs/guides/kafka-quickstart.mdx b/docs/guides/kafka-quickstart.mdx index c5629a99..343f5cf9 100644 --- a/docs/guides/kafka-quickstart.mdx +++ b/docs/guides/kafka-quickstart.mdx @@ -73,20 +73,18 @@ This way, you can use Restate to process events in a lightweight, flexible, tran - Now, let's start the Restate Server and let it know about the Kafka cluster via the configuration file. + Now, let's start the Restate Server: - Store the following configuration in a file named `restate.toml`: - - ```toml restate.toml - [[ingress.kafka-clusters]] - name = "my-cluster" - brokers = ["PLAINTEXT://localhost:9092"] + ```shell + restate-server ``` + + - Start the Restate Server from the same location as the configuration file: + Let the Restate Server know about the Kafka cluster by registering it via the CLI: ```shell - restate-server --config-file restate.toml + restate kafka-clusters create my-cluster bootstrap.servers=localhost:9092 ``` @@ -101,19 +99,15 @@ This way, you can use Restate to process events in a lightweight, flexible, tran Now, we need to make Restate subscribe to the Kafka topics and tell it where it should push the events that arrive on the topic. - Execute the following curl command to create a subscription, and invoke the handler for each event: + Execute the following command to create a subscription, and invoke the handler for each event: ```shell - curl localhost:9070/subscriptions --json '{ - "source": "kafka://my-cluster/greetings", - "sink": "service://Greeter/greet", - "options": {"auto.offset.reset": "earliest"} - }' + restate subscriptions create kafka://my-cluster/greetings service://Greeter/greet auto.offset.reset=earliest ``` For Go, you need to capitalize the handler name: `service://Greeter/Greet`. - This curl command calls the Admin API of the Restate Server and tells it to invoke the `greet` handler of the `Greeter` service for each event that arrives on the `greetings` topic in the `my-cluster` Kafka cluster. + This command tells Restate to invoke the `greet` handler of the `Greeter` service for each event that arrives on the `greetings` topic in the `my-cluster` Kafka cluster. @@ -150,28 +144,16 @@ This way, you can use Restate to process events in a lightweight, flexible, tran - You can see the subscriptions that are active via the Admin API: + You can see the subscriptions that are active via the CLI: ```shell - curl localhost:9070/subscriptions + restate subscriptions list ``` Example output: - ```json - { - "subscriptions": [ - { - "id": "sub_11XHoawrCiWtv8kzhEyGtsR", - "source": "kafka://my-cluster/my-topic", - "sink": "service://Greeter/greet", - "options": { - "auto.offset.reset": "earliest", - "client.id": "restate", - "group.id": "sub_11XHoawrCiWtv8kzhEyGtsR" - } - } - ] - } + ```text + ID SOURCE SINK OPTIONS + sub_11XHoawrCiWtv8kzhEyGtsR kafka://my-cluster/greetings service://Greeter/greet 3 ``` As you can see, subscriptions have an ID that starts with `sub_`. @@ -179,7 +161,7 @@ This way, you can use Restate to process events in a lightweight, flexible, tran Now you can use the subscription ID to delete the subscription: ```shell - curl -X DELETE localhost:9070/subscriptions/sub_11XHoawrCiWtv8kzhEyGtsR + restate subscriptions delete sub_11XHoawrCiWtv8kzhEyGtsR ``` diff --git a/docs/img/monitoring/tracing_invocation_spans.png b/docs/img/monitoring/tracing_invocation_spans.png new file mode 100644 index 00000000..e8d4fbc4 Binary files /dev/null and b/docs/img/monitoring/tracing_invocation_spans.png differ diff --git a/docs/img/monitoring/tracing_span_tags.png b/docs/img/monitoring/tracing_span_tags.png deleted file mode 100644 index aa84325b..00000000 Binary files a/docs/img/monitoring/tracing_span_tags.png and /dev/null differ diff --git a/docs/img/monitoring/tracing_tour.png b/docs/img/monitoring/tracing_tour.png deleted file mode 100644 index 97443ea4..00000000 Binary files a/docs/img/monitoring/tracing_tour.png and /dev/null differ diff --git a/docs/quickstart.mdx b/docs/quickstart.mdx index 9d46bd72..6001fe7b 100644 --- a/docs/quickstart.mdx +++ b/docs/quickstart.mdx @@ -374,7 +374,7 @@ export default { When using [Restate Cloud](https://restate.dev/cloud), your service must be accessible over the public internet so Restate can invoke it. - If you want to develop with a local service, you can expose it using our [tunnel](/cloud/connecting-services#connecting-services-in-private-environments) feature. + If you want to develop with a local service, you can expose it using our [tunnel](/cloud/connecting-services#services-in-private-environments) feature. diff --git a/docs/server/monitoring/tracing.mdx b/docs/server/monitoring/tracing.mdx index 5681ee07..cf40c149 100644 --- a/docs/server/monitoring/tracing.mdx +++ b/docs/server/monitoring/tracing.mdx @@ -5,9 +5,10 @@ description: "Export OTEL traces of your invocations." Restate supports the following tracing features: -* Runtime execution tracing per invocation +* Runtime execution tracing per invocation, exported in real time while the invocation is running * Exporting traces to OTLP-compatible systems (e.g. Jaeger) -* Correlating parent traces of incoming HTTP requests, using the [W3C TraceContext](https://github.com/w3c/trace-context) specification. +* Correlating parent traces of incoming HTTP requests, using the [W3C TraceContext](https://github.com/w3c/trace-context) specification +* Propagating the trace context to your services, so spans created in your handlers join the same trace (see [End-to-end tracing with the SDKs](#end-to-end-tracing-with-the-sdks)) ## Setting up OTLP exporter @@ -38,7 +39,6 @@ restate-server --tracing-endpoint otlp+http://localhost:4318/v1/traces # for HTT If you run Restate in Docker, then instead add the environment variable `-e RESTATE_TRACING_ENDPOINT=http://host.docker.internal:4317`. If you now spin up your services and send requests to them, you will see the traces appear in the Jaeger UI at http://localhost:16686 - @@ -64,27 +64,86 @@ If you now spin up your services and send requests to them, you will see the tra You can import the trace files using the Jaeger UI: - + Import trace files in the Jaeger UI ## Understanding traces -The traces contain detailed information about the context calls that were done during the invocation (e.g. sleep, one-way calls, interaction with state): - +Restate traces represent what is physically happening during an invocation, while it is happening. +For every invocation, Restate emits the following spans: -The initial `ingress_invoke` spans show when the HTTP request was received by Restate. The `invoke` span beneath it shows when Restate invoked the service deployment to process the request. +| Span | Description | +|------|-------------| +| `ingress ` | The HTTP request was received by the Restate ingress. Emitted only for invocations made over the ingress. | +| `invocation-start ` | The invocation started. This is the anchor span: all other spans of the invocation are children of it. | +| `invocation-attempt ` | One span per invocation attempt, emitted as soon as the attempt ends. Attempts that fail with a retryable error are marked with error status, so you can spot retry loops at a glance. | +| `invocation-end ` | The invocation completed, recording whether it succeeded or failed. | -The tags of the spans contain the metadata of the context calls (e.g. call arguments, invocation id). +Spans are exported **as soon as they end**, not when the whole invocation completes. +This means you can inspect invocations that are still running, for example to debug an invocation stuck in a retry loop. - - When a service invokes another service, the child invocation is linked automatically to the parent invocation, as you can see in the image. - Note that the spans of one-way calls are shown as separate traces. The parent invocation only shows that the one-way call was scheduled, not its entire tracing span. - To see this information, search for the trace of the one-way call by filtering on the invocation id tag `restate.invocation.id="inv_19maBIcE9uRD0gIu30mu6eqhZ4pQT"`. +Jaeger UI showing an invocation trace with the start span, multiple attempt spans (failed attempts marked in red), SDK spans, and the end span + +The example above shows an invocation that was retried a few times: each failed attempt is shown as a red `invocation-attempt` span, published right when the attempt failed. +The spans below each attempt are emitted by the service itself, using the [SDK tracing integrations](#end-to-end-tracing-with-the-sdks). + +Operations performed by the handler (e.g. `ctx.run`, calls, sleeps, state access) are recorded as **events on the attempt span**, rather than as separate spans: + +| Event | Description | +|-------|-------------| +| `restate.invocation.lifecycle.new_command` | The handler created a new journal command. The attributes `restate.journal.command.type` and `restate.journal.command.name` describe the command. | +| `restate.invocation.lifecycle.run_ended` | A `ctx.run` block finished executing. | +| `restate.invocation.lifecycle.suspended` | The invocation suspended, waiting for some condition (e.g. a timer, a call result). | +| `restate.invocation.lifecycle.yielded` | The invocation yielded the execution. | + + + When a handler calls another service, the child invocation joins the same trace. + One-way calls (send) are shown as separate traces instead, linked to the trace of the parent invocation. + To find them, search for the trace of the one-way call by filtering on the invocation id attribute `restate.invocation.id`. + + Spans emitted by Restate are exported with the resource service name `Restate`, the process that physically produces them. + The logical, per-invocation view remains available in the Restate UI. + + +### Span attributes + +Restate spans carry the following attributes, which you can use to build dashboards, alerts, and queries: + +| Attribute | Spans | Description | +|-----------|-------|-------------| +| `restate.invocation.id` | All | The invocation ID. | +| `restate.invocation.target` | All | The invocation target (e.g. `Greeter/greet` for services, `Greeter/myKey/greet` for keyed services). | +| `rpc.service` / `rpc.method` | All | The service name and the handler name. | +| `restate.deployment.id` | `invocation-attempt` | The ID of the deployment processing the attempt. | +| `restate.deployment.address` | `invocation-attempt` | The address of the deployment processing the attempt. | +| `restate.deployment.service_protocol_version` | `invocation-attempt` | The service protocol version used by the deployment. | +| `restate.invocation.result` | `invocation-end` | The invocation result: `success` or `failure`. | +| `restate.invocation.error.code` | `invocation-end` | The error code, if the invocation failed. | +| `error.message` | `invocation-end` | The error message, if the invocation failed. | + +## End-to-end tracing with the SDKs + +Restate propagates the [W3C TraceContext](https://github.com/w3c/trace-context) to your service on every invocation attempt. +The SDK tracing integrations use it to create a span per handler attempt, and a child span per `ctx.run` block, all joining the same trace: + +* [TypeScript SDK tracing](/develop/ts/tracing) +* [Java/Kotlin SDK tracing](/develop/java/tracing) + +The SDK spans carry the same `restate.invocation.id` and `restate.invocation.target` attributes as the runtime spans, so you can correlate them easily. +Spans created per `ctx.run` block carry the run name in the `restate.run.name` attribute. + +Trace context propagation also works at the boundaries: + +* **Upstream**: if the incoming HTTP request to the ingress carries a `traceparent` header, the invocation trace continues from it. +* **Downstream**: spans you create yourself inside your handlers (e.g. instrumented HTTP clients, database calls) attach to the trace of the current attempt. + ## Searching traces -Traces export attributes and tags that correlate the trace with the service and/or invocation. For example, in the Jaeger UI, you can filter on the invocation id (`restate.invocation.id`) or any other tag: +Traces export attributes that correlate the trace with the service and/or invocation. For example, in the Jaeger UI, you can filter on the invocation id (`restate.invocation.id`) or any other attribute: + +Searching traces by invocation id in the Jaeger UI - +This also lets you navigate between the Restate UI and your tracing system: copy the invocation ID from the Restate UI and search for it in your tracing system, or vice versa. diff --git a/docs/server/snapshots.mdx b/docs/server/snapshots.mdx index 2761521c..a4f873b7 100644 --- a/docs/server/snapshots.mdx +++ b/docs/server/snapshots.mdx @@ -217,7 +217,7 @@ This gives the snapshot repository time to replicate snapshots to other regions Each partition leader runs a **durability tracker** that monitors: - **Durable LSN**: The log position that has been flushed to local storage on each replica (partition store flush) -- **Archived LSN**: The log position of the latest published snapshot in the object store (or the oldest retained snapshot if `worker.snapshots.experimental-num-retained` is configured) +- **Archived LSN**: The log position of the latest published snapshot in the object store (or the oldest retained snapshot if `worker.snapshots.num-retained` is greater than 1) Based on the configured durability mode, the tracker calculates the **durability point**: the LSN up to which the partition store state is considered safely persisted. Once determined: @@ -262,27 +262,29 @@ There are three notable persistence-related attributes in `restatectl`'s partiti - **Applied LSN** - the latest log record record applied by this processor - **Durable LSN** - the log position of the latest partition store flushed to local node storage; by default processors optimize performance by relying on Bifrost for durability and only periodically flush partition store to disk -- **Archived LSN** - if a snapshot repository is configured, this LSN represents the latest published snapshot (or the oldest retained snapshot if `worker.snapshots.experimental-num-retained` is configured); this determines the log safe trim point in multi-node clusters +- **Archived LSN** - if a snapshot repository is configured, this LSN represents the latest published snapshot (or the oldest retained snapshot if `worker.snapshots.num-retained` is greater than 1); this determines the log safe trim point in multi-node clusters ### Snapshot retention -By default, Restate adds new snapshots without removing old ones. You can configure automatic pruning using the experimental `experimental-num-retained` option: +Restate automatically prunes old snapshots so that snapshot storage does not grow unboundedly. The number of snapshots to retain per partition is controlled by `worker.snapshots.num-retained`, which defaults to `1`: ```toml [worker.snapshots] -experimental-num-retained = 1 +num-retained = 1 ``` -This keeps only the most recent snapshot and automatically deletes older ones. - - - This feature is only available in Restate v1.6 and newer. Only newly uploaded snapshots after the experimental feature was activated will be pruned. Existing snapshots predating the configuration change will not be affected. - +By default, Restate keeps exactly one snapshot per partition and deletes older snapshots once a newer one has been published. Set `num-retained` to a higher value for added resiliency against corrupted snapshots. - When `experimental-num-retained` is greater than 1, the archived LSN advances to the *oldest* retained snapshot rather than the latest. This delays log trimming and increases storage usage on log servers. For most deployments, `experimental-num-retained = 1` is recommended unless you need the ability to fall back to older snapshots. + Restate considers the archived LSN to be that of the *oldest* retained snapshot. When you retain multiple snapshots, you gain the ability to fall back to an older one in case the most recent is corrupted. However, be mindful that this causes an increase in storage usage on the log servers as log trimming will follow the oldest snapshot tracked by Restate. For most deployments, `num-retained = 1` (default) is recommended. + + Snapshot retention was previously opt-in via the `experimental-num-retained` key. Any configuration that still sets `experimental-num-retained` must be renamed to `num-retained` before upgrading; Restate ignores the old key. No action is required for existing snapshot repositories: the first upload after the upgrade rewrites `latest.json` from V1 to V2 in place, and snapshot data files are unchanged. Rolling back to a Restate binary older than v1.6 is not supported once V2 pointers have been written; rolling back to v1.6.x remains safe. + + Note that of the snapshots produced by the older version, only the most recent will be considered for pruning following the upgrade. Older snapshots must still be cleaned up by the user, just as they would have been under v1.6 and earlier. + + ## Data Backups diff --git a/docs/services/deploy/cloud-run.mdx b/docs/services/deploy/cloud-run.mdx new file mode 100644 index 00000000..54f8fcd5 --- /dev/null +++ b/docs/services/deploy/cloud-run.mdx @@ -0,0 +1,104 @@ +--- +title: "Google Cloud Run" +description: "Run your Restate services on Google Cloud Run with native OIDC authentication" +--- + +Deploy your Restate services as containers on Google Cloud Run. Restate can invoke private Cloud Run services natively by minting short-lived, Google-signed OIDC ID tokens for each discovery and invocation request. + +## Set up your service on Cloud Run + +Package your Restate service as a container image and deploy it to Cloud Run following the [Cloud Run documentation](https://cloud.google.com/run/docs/deploying). A typical Restate service exposes its handler on port `9080`; configure the Cloud Run service to forward traffic to whichever port your container listens on. + +Restate's service deployment protocol uses HTTP/2 (h2c), so the Cloud Run service must use end-to-end HTTP/2 to the container. Enable it when you deploy with `--use-http2` (equivalently, the `run.googleapis.com/use-http2` annotation); without it, discovery and invocation fail with an upstream `protocol error`. + +```shell +gcloud run deploy my-restate-service \ + --image \ + --port 9080 \ + --use-http2 \ + --no-allow-unauthenticated +``` + +For a private Cloud Run service, leave the default access policy in place (no unauthenticated invocations). Restate authenticates to the service using a Google OIDC ID token, which Cloud Run validates before forwarding the request to your container. + +## Register the service with native authentication + +When registering an HTTP deployment, opt into native Google OIDC authentication with `--gcp-id-token`. Restate mints a Google-signed ID token for every discovery and invocation request and attaches it as `X-Serverless-Authorization: Bearer `. Cloud Run validates this header in precedence over `Authorization` and strips it before forwarding the request to your container, so any `Authorization` value you set via `--extra-header` reaches the workload unchanged. + +```shell +restate deployments register https://svc-abc-uc.a.run.app --gcp-id-token +``` + +To call the Cloud Run service as a specific service account, impersonate it via the IAM Credentials `generateIdToken` API: + +```shell +restate deployments register https://svc-abc-uc.a.run.app \ + --gcp-impersonate-service-account caller@my-proj.iam.gserviceaccount.com +``` + +`--gcp-impersonate-service-account` implies `--gcp-id-token`. + +### Custom domains and audience overrides + +By default, Restate derives the OIDC `aud` claim from the deployment URL origin (scheme, host, and optional port). If your Cloud Run service sits behind a custom domain, a load balancer, or a traffic tag, set the audience explicitly to the canonical Cloud Run service URL: + +```shell +restate deployments register https://api.acme.com/svc \ + --gcp-audience https://svc-abc-uc.a.run.app +``` + +`--gcp-audience` also implies `--gcp-id-token`. + +### Updating or removing authentication + +Authentication configuration is set when the deployment is registered. To rotate the impersonation target, change the audience, or remove authentication entirely, re-register with `--force`: + +```shell +# Change the impersonation target: +restate deployments register https://svc-abc-uc.a.run.app --force \ + --gcp-impersonate-service-account new-caller@my-proj.iam.gserviceaccount.com + +# Remove authentication: +restate deployments register https://svc-abc-uc.a.run.app --force +``` + +Re-registration with `--force` re-runs discovery against the deployment, so the new configuration must itself be able to reach the service. Removing authentication from a service that still requires authenticated invocations therefore fails at discovery with `403 Forbidden`; only remove authentication once the service accepts the request without a token. + +## Credentials available to Restate + +Restate discovers credentials through Google's [Application Default Credentials (ADC)](https://cloud.google.com/docs/authentication/application-default-credentials) chain. Not every ADC source can mint an OIDC ID token directly; the table below summarizes what works. + +| ADC source | Ambient `--gcp-id-token` | With `--gcp-impersonate-service-account` | +|------------|--------------------------|------------------------------------------| +| GCE / GKE / Cloud Run metadata server | Supported | Supported | +| Service-account JSON key file | Supported | Supported | +| Workload Identity Federation (`external_account`) | Not supported | Supported | +| User credentials from `gcloud auth application-default login` (`authorized_user`) | Not supported | Supported | + +When the ambient identity cannot mint ID tokens for arbitrary audiences, pair it with `--gcp-impersonate-service-account` so Restate calls the IAM Credentials `generateIdToken` API on a target service account that the ambient identity is authorized to impersonate. + +### IAM roles + +The principal whose identity ends up in the minted token must hold: + +- `roles/run.invoker` on the target Cloud Run service, so that Cloud Run IAM accepts the request. +- `roles/iam.serviceAccountOpenIdTokenCreator` on the impersonation target, when `--gcp-impersonate-service-account` is used. + +### Self-hosted Restate + +For self-hosted Restate calling private Cloud Run services, ensure one of the following: + +- The Restate process can reach Google's metadata server (typical on GCE, GKE, or Cloud Run hosts), or +- `GOOGLE_APPLICATION_CREDENTIALS` points to a service-account JSON key file readable by the Restate user. + +Restate does not contact any Google API or read any ADC source until the first deployment with native authentication enabled is observed. + +### Local development + +For local development, run `gcloud auth application-default login` to populate ADC. This is distinct from `gcloud auth login`, which only populates user credentials for the `gcloud` tool itself. + +Authorized-user ADC credentials cannot mint ID tokens for arbitrary audiences without impersonation, so pair them with `--gcp-impersonate-service-account` for the dev flow. + +## Token caching and rotation + +Tokens are minted on demand and cached per `(impersonate_service_account, audience)` pair on the invoker. Cached tokens are evicted 60 seconds before expiry so that every outbound request carries a token with sufficient lifetime. Discovery requests do not cache tokens; a fresh token is minted for each discovery call. diff --git a/docs/services/deploy/kubernetes.mdx b/docs/services/deploy/kubernetes.mdx index 4bd1e54c..67285f01 100644 --- a/docs/services/deploy/kubernetes.mdx +++ b/docs/services/deploy/kubernetes.mdx @@ -72,7 +72,7 @@ The Restate Operator can help you connect your services to Restate Cloud by mana Try connecting your Kubernetes services to Restate Cloud by following this [guide](/guides/connecting-k8s-services-to-cloud). - Or check out the [Restate Cloud documentation](/cloud/connecting-services#connecting-kubernetes-services) for more information. + Or check out the [Restate Cloud documentation](/cloud/connecting-services#kubernetes-services) for more information. ### Automatic Service Versioning diff --git a/docs/services/invocation/kafka.mdx b/docs/services/invocation/kafka.mdx index b33b5943..f266a194 100644 --- a/docs/services/invocation/kafka.mdx +++ b/docs/services/invocation/kafka.mdx @@ -36,57 +36,53 @@ Since you can invoke any handler via Kafka events, a single handler can be invok - -Define the Kafka cluster that Restate needs to connect to in the [Restate configuration file](/server/configuration#configuration-file): + +Register the Kafka cluster that Restate needs to connect to, using the [Restate CLI](/installation): -```toml restate.toml -[[ingress.kafka-clusters]] -name = "my-cluster" -brokers = ["PLAINTEXT://broker:9092"] +```bash +restate kafka-clusters create my-cluster bootstrap.servers=broker:9092 ``` -And make sure the Restate Server uses it via `restate-server --config-file restate.toml`. - -Check the [configuration docs](/server/configuration) for more details. +You can pass any [librdkafka configuration](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) parameter as additional `key=value` arguments. +Alternatively, you can read the properties from a file with `-f my-cluster.properties`, or open an editor on a properties template with `--edit`. -To connect to a Kafka cluster that requires SASL/SSL authentication (e.g., Confluent Kafka), you can specify the necessary parameters in the `restate.toml` file: - -```toml restate.toml -[[ingress.kafka-clusters]] -name = "my-kafka" -brokers = [ "my-kafka:9092" ] -"security.protocol" = "SASL_SSL" -"sasl.mechanisms" = "PLAIN" -"sasl.username" = "user" -"sasl.password" = "pass" - -"client.id" = "client-id" +To connect to a Kafka cluster that requires SASL/SSL authentication (e.g., Confluent Kafka), you can specify the necessary parameters when registering the cluster: + +```bash +restate kafka-clusters create my-kafka \ + bootstrap.servers=my-kafka:9092 \ + security.protocol=SASL_SSL \ + sasl.mechanisms=PLAIN \ + sasl.username=user \ + sasl.password=pass \ + client.id=client-id ``` -Note the quotation marks around the configuration keys. +For Confluent Cloud, you can copy the client configuration properties from the Confluent Cloud console into a properties file and register the cluster with it: -For Confluent Cloud, the rest of the configuration can be copied from the Confluent Cloud "Rust client" configuration. +```bash +restate kafka-clusters create my-kafka -f confluent-cloud.properties +``` The Kafka ingress supports SASL OAUTHBEARER authentication, enabling OAuth 2.0/OpenID Connect (OIDC) token-based connections to managed Kafka services. -Configure SASL OAUTHBEARER via the `additional_options` field in your Kafka cluster configuration. These options are passed directly to librdkafka. +Configure SASL OAUTHBEARER via the cluster properties. These options are passed directly to librdkafka. **Example for Confluent Cloud:** -```toml -[[ingress.kafka-clusters]] -name = "my-cluster" -brokers = ["SASL_SSL://pkc-xxxxxx.eu-central-1.aws.confluent.cloud:9092"] -"security.protocol"="SASL_SSL" -"sasl.mechanisms"="OAUTHBEARER" -"sasl.oauthbearer.method"="oidc" -"sasl.oauthbearer.client.id"="" -"sasl.oauthbearer.client.secret"="" -"sasl.oauthbearer.token.endpoint.url"="" -"sasl.oauthbearer.scope"="kafka" +```bash +restate kafka-clusters create my-cluster \ + bootstrap.servers=pkc-xxxxxx.eu-central-1.aws.confluent.cloud:9092 \ + security.protocol=SASL_SSL \ + sasl.mechanisms=OAUTHBEARER \ + sasl.oauthbearer.method=oidc \ + sasl.oauthbearer.client.id= \ + sasl.oauthbearer.client.secret= \ + sasl.oauthbearer.token.endpoint.url= \ + sasl.oauthbearer.scope=kafka ``` **Common OAUTHBEARER options:** @@ -103,38 +99,6 @@ brokers = ["SASL_SSL://pkc-xxxxxx.eu-central-1.aws.confluent.cloud:9092"] For the full list of available options, see the [librdkafka CONFIGURATION.md](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). - - - -You can also configure the Kafka clusters via the `RESTATE_INGRESS__KAFKA_CLUSTERS` environment variable: - -```bash -RESTATE_INGRESS__KAFKA_CLUSTERS=[{name="my-cluster",brokers=["PLAINTEXT://broker:9092"]}] -``` - - - - In certain scenarios, such as consuming from a Kafka topic with few partitions, the new batch ingestion can significantly improve ingestion throughput compared to the legacy implementation. - - **This feature is disabled by default and should be used with caution.** - - All nodes in the cluster must be running Restate v1.6 before enabling this feature. - - **Once enabled and data has been ingested, you cannot roll back to a version prior to Restatev1.6.** - - To enable the experimental Kafka batch ingestion, set the following environment variable on all nodes: - - ```bash - RESTATE_EXPERIMENTAL_KAFKA_BATCH_INGESTION=true - ``` - - Or in your configuration file: - - ```toml - experimental-kafka-batch-ingestion = true - ``` - - Contact us on [Discord](https://discord.restate.dev) or [Slack](https://slack.restate.dev) to test it together with us. @@ -144,17 +108,13 @@ RESTATE_INGRESS__KAFKA_CLUSTERS=[{name="my-cluster",brokers=["PLAINTEXT://broker Let Restate forward events from the Kafka topic to the event handler by creating a subscription: ```bash -curl localhost:9070/subscriptions --json '{ - "source": "kafka://my-cluster/my-topic", - "sink": "service://MyService/handle", - "options": {"auto.offset.reset": "earliest"} -}' +restate subscriptions create kafka://my-cluster/my-topic service://MyService/handle auto.offset.reset=earliest ``` Once you've created a subscription, Restate immediately starts consuming events from Kafka. The handler will be invoked for each event received from Kafka. -The `options` field is optional and accepts any configuration parameter from [librdkafka configuration](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). +The trailing `key=value` options are optional and accept any configuration parameter from [librdkafka configuration](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). @@ -162,14 +122,13 @@ The `options` field is optional and accepts any configuration parameter from [li -You can pass arbitrary Kafka cluster options in the `restate.toml`, and those options will be applied for all the subscriptions to that cluster, for example: +You can pass arbitrary Kafka cluster properties when registering the cluster, and those properties will be applied for all the subscriptions to that cluster, for example: -```toml restate.toml -[[ingress.kafka-clusters]] -name = "my-cluster" -brokers = ["PLAINTEXT://broker:9092"] -"sasl.username" = "me" -"sasl.password" = "pass" +```bash +restate kafka-clusters create my-cluster \ + bootstrap.servers=broker:9092 \ + sasl.username=me \ + sasl.password=pass ``` For the full list of options, check [librdkafka configuration](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). @@ -177,32 +136,21 @@ For the full list of options, check [librdkafka configuration](https://github.co -You can configure multiple kafka clusters in the `restate.toml` file: - -```toml restate.toml -[[ingress.kafka-clusters]] -name = "my-cluster-1" -brokers = ["PLAINTEXT://localhost:9092"] +You can register multiple Kafka clusters: -[[ingress.kafka-clusters]] -name = "my-cluster-2" -brokers = ["PLAINTEXT://localhost:9093"] +```bash +restate kafka-clusters create my-cluster-1 bootstrap.servers=localhost:9092 +restate kafka-clusters create my-cluster-2 bootstrap.servers=localhost:9093 ``` -And then, when creating the subscriptions, you refer to the specific cluster by `name`: +And then, when creating the subscriptions, you refer to the specific cluster by name: ```bash # Subscription to my-cluster-1 -curl localhost:9070/subscriptions --json '{ - "source": "kafka://my-cluster-1/topic-1", - "sink": "service://MyService/handleCluster1" -}' +restate subscriptions create kafka://my-cluster-1/topic-1 service://MyService/handleCluster1 # Subscription to my-cluster-2 -curl localhost:9070/subscriptions --json '{ - "source": "kafka://my-cluster-2/topic-2", - "sink": "service://MyService/handleCluster2" -}' +restate subscriptions create kafka://my-cluster-2/topic-2 service://MyService/handleCluster2 ``` @@ -227,7 +175,7 @@ ctx.request().headers Each event carries within this map the following entries: -* `restate.subscription.id`: The subscription identifier, as shown by the Admin API. +* `restate.subscription.id`: The subscription identifier, as shown by `restate subscriptions list`. * `kafka.offset`: The record offset. * `kafka.partition`: The record partition. * `kafka.timestamp`: The record timestamp. @@ -238,56 +186,36 @@ Check out the serialization documentation of your SDK to learn how to receive ra -## Managing Kafka Subscriptions - -Restate can trigger handlers via Kafka events. - -### Create Subscriptions +## Managing Kafka Clusters -Subscribe a handler to a Kafka topic: +Manage the registered Kafka clusters with the `restate kafka-clusters` CLI commands (alias `kc`): ```bash -curl localhost:9070/subscriptions --json '{ - "source": "kafka://my-cluster/my-topic", - "sink": "service://MyService/Handle", - "options": {"auto.offset.reset": "earliest"} -}' +# List the registered Kafka clusters +restate kafka-clusters list +# Print the properties of a Kafka cluster and its subscriptions +restate kafka-clusters describe my-cluster +# Open an editor to interactively update the cluster properties +restate kafka-clusters edit my-cluster +# Update the cluster properties non-interactively, for CI and scripting +restate kafka-clusters patch my-cluster --set client.id=restate-cli --unset sasl.password +# Remove a Kafka cluster +restate kafka-clusters delete my-cluster ``` -The `options` field is optional and accepts any [librdkafka configuration](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) parameter. - -### List Subscriptions - -View current subscriptions: - -```bash -curl localhost:9070/subscriptions -``` - -**Example response:** -```json -{ - "subscriptions": [ - { - "id": "sub_11XHoawrCiWtv8kzhEyGtsR", - "source": "kafka://my-cluster/my-topic", - "sink": "service://Greeter/greet", - "options": { - "auto.offset.reset": "earliest", - "client.id": "restate", - "group.id": "sub_11XHoawrCiWtv8kzhEyGtsR" - } - } - ] -} -``` - -### Delete Subscriptions +## Managing Kafka Subscriptions -Remove a subscription using its ID (starts with `sub_`): +Manage the subscriptions with the `restate subscriptions` CLI commands (alias `sub`): ```bash -curl -X DELETE localhost:9070/subscriptions/sub_11XHoawrCiWtv8kzhEyGtsR +# Subscribe a handler to a Kafka topic +restate subscriptions create kafka://my-cluster/my-topic service://MyService/Handle auto.offset.reset=earliest +# List current subscriptions +restate subscriptions list +# Print detailed information about a subscription, including its options +restate subscriptions describe sub_11XHoawrCiWtv8kzhEyGtsR +# Remove a subscription using its ID (starts with sub_) +restate subscriptions delete sub_11XHoawrCiWtv8kzhEyGtsR ``` When you delete a subscription, Restate stops the associated consumer group. Messages already enqueued by Restate will still be processed.