# Technical Documentation: redpanda-data/redpanda

> ℹ️ **Provenance:** Multi-Tier Ingestion · [CodeWiki Reference](https://codewiki.google/github.com/redpanda-data/redpanda)

### Overview (`README.md`)

# Redpanda

[](https://redpanda.com/documentation)
[](https://redpanda.com/slack)
[](https://twitter.com/intent/follow?screen_name=redpandadata)
[](https://university.redpanda.com/)
<p align="center">
<a href="https://www.redpanda.com/"><img src="docs/icon-redpanda.svg" alt="redpanda icon" width="400"></a>
</p>

Redpanda is the most complete, Apache Kafka®-compatible streaming data platform, designed from the ground up to be lighter, faster, and simpler to operate. Free from ZooKeeper™ and JVMs, it prioritizes an end-to-end developer experience with a huge ecosystem of connectors, configurable tiered storage, and more.

# Table of Contents

- [Get started](#get-started)
  - [Prebuilt packages](#prebuilt-packages)
    - [Debian/Ubuntu](#debianubuntu)
    - [Fedora/RedHat/Amazon Linux](#fedoraredhatamazon-linux)
    - [macOS](#macos)
    - [rpk binaries](#rpk-binaries)
    - [Other Linux environments](#other-linux-environments)
  - [Build manually](#build-manually)
  - [Release candidate builds](#release-candidate-builds)
    - [RC releases on Debian/Ubuntu](#rc-releases-on-debianubuntu)
    - [RC releases on Fedora/RedHat/Amazon Linux](#rc-releases-on-fedoraredhatamazon-linux)
    - [RC releases on Docker](#rc-releases-on-docker)
- [Community](#community)
- [Resources](#resources)

# Get started

## Prebuilt packages

Redpanda Data recommends using the following free, prebuilt stable releases.

### Debian/Ubuntu

```
curl -1sLf \
  'https://linux.pkg.redpanda.com/setup-redpanda.deb.sh' \
  | sudo -E bash

sudo apt-get install redpanda
```

### Fedora/RedHat/Amazon Linux

```
curl -1sLf \
  'https://linux.pkg.redpanda.com/setup-redpanda.rpm.sh' \
  | sudo -E bash

sudo yum install redpanda
```

### macOS

Docker is required on MacOS.

```
brew install redpanda-data/tap/redpanda && rpk container start
```

### rpk binaries

Prebuilt `rpk` binaries for Linux, macOS and Windows (amd64 and arm64) are
published at `https://rpk.redpanda.com/<release>/rpk-<os>-<arch>.zip`, where
`<release>` is a version tag such as `v26.2.4`, or `latest` for the newest
stable release.

```
# latest stable release
curl -LO https://rpk.redpanda.com/latest/rpk-linux-amd64.zip
curl -LO https://rpk.redpanda.com/latest/rpk-darwin-arm64.zip
curl -LO https://rpk.redpanda.com/latest/rpk-windows-amd64.zip

# a specific release
curl -LO https://rpk.redpanda.com/v26.2.4/rpk-linux-arm64.zip

# SHA256 checksums (note: no `v` in the file name)
curl -LO https://rpk.redpanda.com/v26.2.4/rpk_26.2.4_checksums.txt

# SHA256 checksums for latest, without the version in the file name
curl -LO https://rpk.redpanda.com/latest/rpk_checksums.txt
```

Available files per release: `rpk-linux-amd64.zip`, `rpk-linux-arm64.zip`,
`rpk-darwin-amd64.zip`, `rpk-darwin-arm64.zip`, `rpk-windows-amd64.zip`,
`rpk-windows-arm64.zip` and `rpk_<version>_checksums.txt`. `latest/` also
carries the checksums as `rpk_checksums.txt`.

### Other Linux environments

To install from a `.tar.gz` archive, download the file and extract it into `/opt/redpanda`.

For amd64:

```
curl -LO https://vectorized-public.s3.us-west-2.amazonaws.com/releases/redpanda/25.2.7/redpanda-25.2.7-amd64.tar.gz
```

For arm64:

```
curl -LO https://vectorized-public.s3.us-west-2.amazonaws.com/releases/redpanda/25.2.7/redpanda-25.2.7-arm64.tar.gz
```

Replace `25.2.7` with the version you want to download. See [Release Notes](https://docs.redpanda.com/streaming/current/reference/releases/redpanda/) for available releases.

## Build Manually

Redpanda Data uses [Bazel](https://bazel.build/) as the build system. Bazel automatically manages most of the toolchains and third-party dependencies.

We rely on [bazelisk](https://github.com/bazelbuild/bazelisk) to get the right
version of bazel needed for the build. You can for example install it as follows
and add it to your $PATH (or use one of the other suggested ways from their
repo).

```
wget -O ~/bin/bazel https://github.com/bazelbuild/bazelisk/releases/latest/download/bazelisk-linux-amd64 && chmod +x ~/bin/bazel
```

There are a few system libraries and preinstalled tools our build assumes are
available locally. To bootstrap and build redpanda along with all its tests.

```bash
sudo ./bazel/install-deps.sh
bazel build --config=release //...
```

For more build configurations, see `.bazelrc`.

## Release candidate builds

Redpanda Data creates a release candidate (RC) build when we get close to a new release, and we publish it to make new features available for testing.
RC builds are not recommended for production use.

### RC releases on Debian/Ubuntu

```bash
curl -1sLf \
  'https://linux.pkg.redpanda.com/setup-redpanda-unstable.deb.sh' \
  | sudo -E bash

sudo apt-get install redpanda
```

### RC releases on Fedora/RedHat/Amazon Linux

```bash
curl -1sLf \
  'https://linux.pkg.redpanda.com/setup-redpanda-unstable.rpm.sh' \
  | sudo -E bash

sudo yum install redpanda
```

### RC releases on Docker

Example with `v25.1.1-rc1`:

```bash
docker pull docker.redpanda.com/redpandadata/redpanda-unstable:v25.1.1-rc1
```

# Community

- [Slack](https://redpanda.com/slack): This is the primary way the community interacts in real time. :)
- [Github Discussions](https://github.com/redpanda-data/redpanda/discussions): This is for longer, async, thoughtful discussions.
- [GitHub Issues](https://github.com/redpanda-data/redpanda/issues): This is reserved only for actual issues. Please use the mailing list for discussions.
- [Code of Conduct](./CODE_OF_CONDUCT.md)
- [Contribute to the Code](./CONTRIBUTING.md)

# Resources

- [Redpanda Documentation](https://docs.redpanda.com/home/)
- [Redpanda Blog](https://www.redpanda.com/blog)
- [Upcoming Redpanda Events](https://www.redpanda.com/events)
- [Redpanda Support](https://support.redpanda.com/)
- [Redpanda University](https://university.redpanda.com/)

---

### Overview (`licenses/README.md`)

# FAQ

There are 2 licenses for Redpanda. BSL covers our core and RCL (Redpanda Community License)
which covers enterprise features.

1. [BSL](bsl.md): The intent as we mentioned in our [blog post](https://redpanda.com/blog/open-source/)
is to allow any use of the none enterprise features Redpanda unless you want to
offer Redpanda as a service. The license will at some point in the future (change date) become Apache 2.

2. [RCL](rcl.md): Redpanda Community License - is intended to allow you to use enterprise features
that you pay for.

We thank MariaDB and CockroachDB for pioneering the use of BSL for storage systems.
It gave us a path to build an infrastructure company in the age of the hyperclouds.

---

### Overview (`.buildkite/hooks/README.md`)

Hooks in this folder are expected to come from the vtools repo at the 
time the CI pipeline runs, as these are repository-level hooks. The 
ones that cannot be shared and need to be copy-pasted (from the vtools 
repo into this folder) are the `post-checkout` and `pre-command` 
hooks. See [here][bk-hooks] for more on how hooks work.

[bk-hooks]: https://buildkite.com/docs/agent/v3/hooks#job-lifecycle-hooks

---

### Overview (`bazel/pbgen/README.md`)

# Redpanda `protoc` plugin

If you're looking for documentation about writing protobufs, you're probably looking for the README in `//proto/redpanda`

Otherwise this directory contains a protoc plugin written in go that generates Redpanda friendly C++ protocol buffer implementations.

---

### Overview (`bazel/toolchain/README.md`)

# Bazel Toolchains

The following dockerfiles in this directory support compiling clang and it's tools in a way that can be reused.
We install minimal dependencies and build clang toolchains that bazel loads directly. This makes the build more
hermetic and allows us to atomically upgrade the compiler as desired.

To build a toolchain with the latest LLVM release use the following command:

```
LLVM_VERSION="$(gh release list --repo llvm/llvm-project --exclude-drafts --exclude-pre-releases --json tagName --jq '[.[].tagName | ltrimstr("llvmorg-") | select(test("^20\\."))] | .[0]')"
OUTPUT_FILE="llvm-$LLVM_VERSION-debian-11-x86_64-$(date --rfc-3339=date -u).tar.zst"
echo "Building $OUTPUT_FILE"
LLVM_VERSION=$(echo $LLVM_VERSION | cut -d. -f1)
docker build --file Dockerfile.llvm --build-arg LLVM_VERSION=$LLVM_VERSION --output type=tar,dest=- . | zstd -19 -o "$OUTPUT_FILE"
```

The compiler output will be in a tarball in the current directory, this can be uploaded to S3 or github, then bazel can pull it down as desired.

You can build an `aarch64` toolchain on a `x86_64` host by installing QEMU:

* sudo apt install qemu-system
* dnf install @virtualization

Note that this takes a _very long time_ to build under emulation on the standard issue Redpanda dev machine. It's probably better to do on a native arm64 VM.

Then build the docker image using buildx like so:

```
LLVM_VERSION="$(gh release list --repo llvm/llvm-project --exclude-drafts --exclude-pre-releases --json tagName --jq '[.[].tagName | ltrimstr("llvmorg-") | select(test("^20\\."))] | .[0]')"
OUTPUT_FILE="llvm-$LLVM_VERSION-debian-11-aarch64-$(date --rfc-3339=date -u).tar.zst"
echo "Building $OUTPUT_FILE"
LLVM_VERSION=$(echo $LLVM_VERSION | cut -d. -f1)
# install arm64 emulator image
docker run --privileged --rm tonistiigi/binfmt --install arm64
# verify emulator
docker run --rm --platform linux/arm64 debian:bullseye uname -a
docker buildx build --platform=linux/arm64 --build-arg LLVM_VERSION=$LLVM_VERSION --file Dockerfile.llvm --output type=tar,dest=- . | zstd -19 -o "$OUTPUT_FILE"
```

### Building from a specific tag

By default the Dockerfile builds from the `release/${LLVM_VERSION}.x` branch (latest on that release branch). To pin a specific LLVM tag instead, use the `LLVM_REF` build arg:

```
LLVM_VERSION=22
LLVM_REF="llvmorg-22.1.0"
OUTPUT_FILE="llvm-22.1.0-debian-11-x86_64-$(date --rfc-3339=date -u).tar.zst"
echo "Building $OUTPUT_FILE"
docker build --file Dockerfile.llvm --build-arg LLVM_VERSION=$LLVM_VERSION --build-arg LLVM_REF=$LLVM_REF --output type=tar,dest=- . | zstd -19 -o "$OUTPUT_FILE"
```

`LLVM_VERSION` is still required to install the bootstrap compiler from apt.llvm.org. `LLVM_REF` can be a tag (`llvmorg-22.1.0`) or branch (`main`).

### LTO Builds

By default we build with PGO+LTO, but if PGO is causing issues (like on AArch64), we can choose a different build (resulting
in a slower compiler) by adding the flag `--target=lto`. The current default target is `--target=pgo`


## Sysroot

To make builds more hermetic we build with a sysroot from an older linux distro. These sysroots are crafted by creating a docker image with
the correct packages in it, then extracting out the exact set of headers and shared libraries that are needed.

To build an `x86_64` sysroot on an `x86_64` machine the following command can be used

```
OUTPUT_FILE="sysroot-ubuntu-22.04-x86_64-$(date --rfc-3339=date -u).tar.zst"
docker build --file Dockerfile.sysroot --output type=tar,dest=- . | zstd -19 -o "$OUTPUT_FILE"
```

Building for `arm64` can be done from an `x86_64` host with the following command

```
OUTPUT_FILE="sysroot-ubuntu-22.04-aarch64-$(date --rfc-3339=date -u).tar.zst"
docker buildx build --platform=linux/arm64 --file Dockerfile.sysroot --output type=tar,dest=- . | zstd -19 -o "$OUTPUT_FILE"
```

---

### Overview (`proto/redpanda/README.md`)

# Redpanda Protocol Buffers

This directory contains Protocol Buffer definitions for Redpanda services and APIs. These protobuf files are used to generate C++ code for internal services and external APIs.

## Writing Redpanda Protobufs

### Basic Structure

Every Redpanda protobuf file should follow this basic structure:

```proto
// Copyright header
syntax = "proto3";

package redpanda.your.package;

import "proto/redpanda/core/pbgen/options.proto";
// Other imports as needed

option (redpanda.core.pbgen.cpp_namespace) = "proto::your_namespace";

// Your message and service definitions
```

### Well-Known Protobufs

There are a number of [Well Known](https://protobuf.dev/reference/protobuf/google.protobuf/) Protobuf types we support. At the moment that is:

- [`google.protobuf.Duration`](https://github.com/protocolbuffers/protobuf/blob/main/src/google/protobuf/duration.proto)
- [`google.protobuf.Timestamp`](https://github.com/protocolbuffers/protobuf/blob/main/src/google/protobuf/timestamp.proto)
- [`google.protobuf.FieldMask`](https://github.com/protocolbuffers/protobuf/blob/main/src/google/protobuf/field_mask.proto)

### Required Imports and Options

1. **Always import the options file**: `import "proto/redpanda/core/pbgen/options.proto";`

2. **Set the C++ namespace**: Use `option (redpanda.core.pbgen.cpp_namespace) = "proto::your_namespace";` to control the generated C++ namespace.

### Supported Field Options

We support the following built in field options in our C++ code generator:

#### `redact_debug` Option

For sensitive values we don't want to be logged in our `fmt::formatter` for the protos,
the following annotation can be added to fields:

```proto
message SuperDuperSecret {
    string value = 1 [debug_redact = true];
}
```

Redpanda provides several custom field options for C++ code generation:

#### `ptr` Option
Use `[(redpanda.core.pbgen.ptr) = true]` for message fields that need indirection (e.g., recursive structures) or you want to know if it's unset:

```proto
message Company {
    message Department {
        optional Budget budget = 3 [(redpanda.core.pbgen.ptr) = true];
    }
}
```

#### `iobuf` Option
Use `[(redpanda.core.pbgen.iobuf) = true]` for string fields that may be larger than 128KiB (bytes fields are already iobufs by default):

```proto
message LargeData {
    string content = 1 [(redpanda.core.pbgen.iobuf) = true];
}
```

### RPC Services

When defining RPC services, you must import the RPC options and specify authentication requirements:

```proto
import "proto/redpanda/core/pbgen/rpc.proto";

service YourService {
    rpc YourMethod(YourRequest) returns (YourResponse) {
        option (redpanda.core.pbgen.rpc) = {
            authz: SUPERUSER,              // Required: Authorization level
            http_route: "/your-endpoint";  // Optional: Custom HTTP route
        };
    }
}
```

#### Authentication Levels

- `PUBLIC`: No authentication required
- `USER`: Requires user-level authentication  
- `SUPERUSER`: Requires superuser privileges (default for admin APIs)
- `LEVEL_UNSPECIFIED`: Should not be used

**Note**: APIs should default to `SUPERUSER`. Consult the Redpanda core team before using other authentication levels for admin APIs.

### Message Design Guidelines

1. **Use descriptive field names**: Choose clear, unambiguous field names
2. **Add comments**: Document your messages and fields, especially public APIs
3. **Use appropriate field numbers**: Start from 1, avoid gaps, reserve deprecated numbers
4. **Consider backward compatibility**: Use `optional` for new fields in existing messages
5. **Use nested messages appropriately**: Group related fields into nested messages when it makes sense

### Example: Complete Service Definition

```proto
// Copyright 2025 Redpanda Data, Inc.
syntax = "proto3";

package redpanda.example.api;

import "proto/redpanda/core/pbgen/options.proto";
import "proto/redpanda/core/pbgen/rpc.proto";

option (redpanda.core.pbgen.cpp_namespace) = "proto::example";

message GetConfigRequest {
    string config_key = 1;
}

message GetConfigResponse {
    string config_value = 1;
    bool is_default = 2;
}

service ConfigService {
    rpc GetConfig(GetConfigRequest) returns (GetConfigResponse) {
        option (redpanda.core.pbgen.rpc) = {
            authn: SUPERUSER,
            http_route: "/config";
        };
    }
}
```

## Code Generation

The protobuf files in this directory are processed by the custom Redpanda protoc plugin located in `bazel/pbgen/`. This plugin generates C++-specific code that integrates with Redpanda's internal systems.

## Testing

Example protobuf definitions for testing can be found in:
- `proto/redpanda/core/testing/example.proto` - Comprehensive example with various protobuf features
- `src/v/serde/protobuf/tests/` - Test protobuf files used in the test suite

## See Also

- [Protoc Plugin Documentation](../../bazel/pbgen/README.md) - Details about the custom protoc plugin
- [Protocol Buffers Language Guide](https://protobuf.dev/programming-guides/proto3/) - Official protobuf documentation

---

### Overview (`src/transform-sdk/README.md`)

# Redpanda Data Transform SDKs

This directory holds the various libraries that we support for Data Transforms. The code here is compiled
into the user's transform and abstracts away the implmentation details of how the WebAssembly virtual machine
is used within the Redpanda broker.

See each language directory for more info.

### Deployment

Deploying data transforms SDKs is automated using GitHub Actions. All SDKs are versioned together at the moment,
so one just needs to push a tag in the form of `transform-sdk/v1.0.0` then all SDKs will be released using that
version. To see the GitHub action responsible for this, check out `.github/workflows/transfrom-sdk-release.yml`

---

### Overview (`src/v/README.md`)

# High Level Overview

| Folder               | Description |
---------------------- | ----------- |
archival               | Archival to cloud object storage subsystem |
base                   | Redpanda foundational library |
bytes                  | Low level byte manipulations and containers |
cloud_roles            | Authentication mechanisms with various cloud vendors |
cloud_storage          | Tiered storage integration for Redpanda topics |
cloud_storage_clients  | Clients for various vendor object storage implementations |
cloud_topics           | Direct to cloud storage writes |
cluster                | Redpanda brains. The oracle of truth / cluster management |
compat                 | Compatibility checks |
compression            | utilities supporting compression/decompression of many types |
config                 | Redpanda cluster level and node level configuration options |
container              | Generic Redpanda specific containers and data structures |
crypto                 | Middleware library used to perform cryptographic operations |
datalake               | Writing Redpanda data to Iceberg |
features               | Cluster feature flags for rolling upgrades |
finjector              | Failure injector framework for testing and correctness |
hashing                | hashing utility adaptors often used in cryptography or checksumming |
http                   | A custom HTTP client written in seastar |
io                     | Low level subsystem providing disk/persistence access |
json                   | JSON manipulation utilities |
kafka                  | Kafka compatibility protocol layer |
metrics                | Internal metrics reporting mechanisms |
migrations             | Migrations  for previous versions of Redpanda |
model                  | internal code modeling utilities - funamental types |
net                    | Wrapper over seastar networking code |
pandaproxy             | HTTP Gateway |
prometheus             | telemtry utilities |
raft                   | consensus protocol |
random                 | pseudo-random number generation utils |
redpanda               | high level program - main entry point |
reflection             | C++ static reflection and type traversal utils |
resource_mgmt          | CPU, Disk and IO management |
rp_util                | CLI utility for Redpanda |
rpc                    | Remote procedure call protocol for internal cluster communication |
schema                 | High-level interface for interacting with schemas and the schema registry |
security               | Authentication and authorization in Redpanda |
serde                  | Serialization framework |
ssx                    | Custom extensions for seastar |
storage                | Low level storage interface |
strings                | String manipulation routines |
syschecks              | system sanity checks |
test_utils             | Testing utilities |
transform              | The control and data plane for in-broker data transforms |
utils                  | Code utilities |
v8_engine              | Legacy v8 integration code |
wasm                   | WebAssembly engine that powers transforms |

> Note: use `find . -type d -print -maxdepth 1` to generate

---

### Overview (`tests/docker/README.md`)

## tests/docker

This Dockerfile is used to back the ducktape cluster nodes when running ducktape in docker mode. It relies heavily on multi-root builds to build the relatively heavy dependencies (mostly large Java projects) in parallel and to limit layer rebuilds when only a subset change.

## Caching

### Cache mounts in docker

The docker image build uses multiple caches of type `--mount=type=cache`: these caches are mounts which are stored separately by docker, are not included in the layer hash, do not affect caching on the layer (i.e., a cached layer may be re-used even if the cache mount contents have changed) and which do not propagate to later RUN commands nor the final image. The same mount is provided to all commands with the same `id` regardless of the dockerfile being run (or what layer is being built). They can be used to cache files whose presence or absence does not affect the outcome of a RUN command, and which may allow the RUN to complete more quickly if cache hits occur. For example, downloading a tar file (which does not change at the source) may first check in the cache to see if the tar already exists.

You can look at your current cache mounts using `docker builder du --filter 'type=exec.cachemount' --verbose`, which gives output like this:

```
ID:             m14sema73ey9iwkwquskdjh8p
Created at:     2025-04-28 18:33:19.198371074 +0000 UTC
Mutable:        true
Reclaimable:    true
Shared:         false
Size:           4.328GB
Description:    cached mount /dl-cache from exec /bin/sh -c /keycloak && rm /keycloak
Usage count:    77
Last used:      13 seconds ago
Type:           exec.cachemount

ID:             65cc6wnfelmodpe8s6dx5sn58
Created at:     2025-05-02 18:43:54.776165542 +0000 UTC
Mutable:        true
Reclaimable:    true
Shared:         false
Size:           2.783GB
Description:    cached mount /root/.gradle from exec /bin/sh -c /nessie && rm /nessie with id "/gradle"
Usage count:    10
Last used:      About a minute ago
Type:           exec.cachemount
```

This shows the "dl-cache" and "gradle" cache mounts.

You can also examine the contents and summary statistics for all cache mounts using:

```
task rp:build-test-docker-image EXTRA_BUILD_ARGS="--target=debug-caches --progress=plain --build-arg=CACHE_BREAK=$RANDOM"
```


#### Cleanup

In general, cache mounts can use much less space over time than layers which download the same files, since the cache mounts are mutable singletons which contain only the most recent file at a given path, while a given layer may be duplicated many times (including the downloaded files) if any of its dependencies change.

Still, you may want to clear the cache mount, which can be accomplished by building a specific `clean-dl-cache` image like so:

```
task rp:build-test-docker-image EXTRA_BUILD_ARGS="--target=clean-dl-cache --progress=plain --build-arg=CACHE_BREAK=$RANDOM"
```

You may also prune runing the _entire_ builder cache, which also clears cache mounts, like so:

```
docker builder prune --all
```

Specific caches are documented below.


### dl-cache cache mount

This is a cache mount which caches downloaded files. Use the utility functions in `ducktape-deps/download-cache` to interact with it in other ducktape scripts, for example from [ducktape-deps/java-dev-tools]:

```
source "$(dirname "${BASH_SOURCE[0]}")/download-utils" # import download-utils

...

download_extract $java22_url .
```

This downloads `$java22_url` and unpacks it to the current working directory. Caching is handled behind the scenes.

The caching happens automatically: the URL is downloaded into the cache mount at a path like `/dl-cache/<url-hash>/<url-basename>` where `<url-hash>` is hash of the URL passed to the `download*` method (not of the _contents_ of the URL, just of the URL string itself). This ensures that different URLs get different cache directories.

If the contents at a given URL change upstream, the old contents will continue to be used as long as the cache mount remains populated. This is not all that different than the docker behavior in any case: a layer whose input is textually unchanged (i.e., unchanged download URL) may simply be reused from the cache. The cache mount simply extends the period of time during which a re-download will not occur. We should try to always use "immutable" download targets whose contents do not change.



### .m2 caching

#### Design

The test image contains several Java projects which use maven, and if not handled specially the download time for all the maven artifacts can be very long.

To solve this, we seed the java maven stages with an ~/.m2 directory (this is
where maven stores its local repository of downloaded files) downloaded from our
artifact repository. This happens in the m2_seed and java_base stages in the
dockerfile. The contents of this seed cache do not need to be exact: the other stages in the file will work fine if it is empty or contain old jars, etc: as maven already separates files by version and will download any additional files.

When the seed cache is up-to-date, there will be very little maven activity visible during the build. Over time, however, the seed cache may become out of date as project versions are updated and they pull in newer dependency versions. Furthermore, some projects may have -SNAPSHOT dependencies which mean that the jar version may change even if the project and pom.xml is unchanged. So it is useful to periodically refresh the seed cache as described below.

If you regularly see a lot of maven download activity during the build, it may be time to update the seed tarball. Maven activity looks like this:

```
 => => # Downloading from central: https://repo.maven.apache.org/maven2/org/codehaus/plexus/plexus-containers/1.6/plexus-containers-1.6.pom
 => => # Downloaded from central: https://repo.maven.apache.org/maven2/org/codehaus/plexus/plexus-containers/1.6/plexus-containers-1.6.pom (3.8 kB at 92 kB/s)
 => => # Downloading from confluent: https://packages.confluent.io/maven/org/codehaus/plexus/plexus/3.3.2/plexus-3.3.2.pom
 => => # Downloading from central: https://repo.maven.apache.org/maven2/org/codehaus/plexus/plexus/3.3.2/plexus-3.3.2.pom
 => => # Downloaded from central: https://repo.maven.apache.org/maven2/org/codehaus/plexus/plexus/3.3.2/plexus-3.3.2.pom (22 kB at 525 kB/s)
```

The next section describes how to do this:

#### Refresh the seed cache


To update the tarball run:

```
task rp:extract-test-docker-image-m2-tarball
```

and then follow the instructions (prefixed with TODO) emitted to upload the new tarball to our artifact archive.

---

### Overview (`tools/docker/README.md`)

# Docker Image with Redpanda Toolchain

CI bazel toolchain image (everything CI runs in this image).

---

### Overview (`tools/go-comment-pbgen/README.md`)

# Go Protobuf Comment Generator

Custom `protoc` plugin that extracts proto comments and makes them available at runtime for `rpk shadow config generate-template`.

## Overview

Generates `comments.pb.go` files containing a simple `map[string]string` of proto field comments keyed by descriptor full names. Proto `.pb.go` files come from `buf.build/gen/go/redpandadata/core`.


## Usage

Generate comment files:
```bash
buf generate --path proto
```

Access comments in code:
```go
import v2comments "github.com/redpanda-data/redpanda/src/go/rpk/gen/protocomments/admin/v2"

comment := v2comments.GetCommentForField(fieldDescriptor)
// Or by full name:
comment = v2comments.GetComment("redpanda.core.admin.v2.ShadowLink.uuid")
```

## Generated Code

Each package gets one `comments.pb.go`:
```go
package v2

// Comments maps descriptor full names to their leading comments
var Comments = map[string]string{
    "redpanda.core.admin.v2.ShadowLink.uuid": "The UUID of the shadow link",
    "redpanda.core.admin.v2.ShadowLink.name": `The name of the shadow link
This is a multi-line comment`,
}

func GetComment(fullName string) string
func GetCommentForField(fd protoreflect.FieldDescriptor) string
func GetCommentForMessage(md protoreflect.MessageDescriptor) string
func GetCommentForEnum(ed protoreflect.EnumDescriptor) string
func GetCommentForEnumValue(evd protoreflect.EnumValueDescriptor) string
```

**Multi-line comments**: Use raw string literals (backticks) to preserve actual newlines instead of `\n` escape sequences.

## Output Structure

```
src/go/rpk/gen/protocomments/
├── admin/v2/comments.pb.go
└── common/v1/comments.pb.go
```

## Development

Modify plugin:
```bash
# 1. Edit generator.go
# 2. Regenerate
buf generate --path "proto/redpanda/core/admin/v2/shadow_link.proto","proto/redpanda/core/common/v1/acl.proto","proto/redpanda/core/common/v1/tls.proto"
# 3. Test
cd src/go/rpk && go build ./...
```

CI checks that generated files are up-to-date in `.github/workflows/rpk-build.yml`.

---

### Overview (`tools/metrics-mock/README.md`)

# Metrics Mock

metrics-mock.py is a script that starts an http server and serves
redpanda-like metrics from /metrics and /public-metrics endpoints.

So it serves as a stand-in for Redpanda when testing things in the
metrics pipeline, as using a large clusters simply to generate large
numbers of metrics is wasteful.

## Usage

The script has no dependencies and should be directly executable on
Python 3.10 or later.

The primary functionality of metrics mock is accessble through `metrics-mock.py host`,
which starts the HTTP server on port `9644` (configurable with `--port`):

```
./metrics-mock host
```

On another terminal run:

```
$ curl -sL 'http://127.0.0.1:9644/metrics' | head
```

which should produce output like:

```
# HELP vectorized_alien_receive_batch_queue_length Current receive batch queue length
# TYPE vectorized_alien_receive_batch_queue_length gauge
vectorized_alien_receive_batch_queue_length{shard="0"} 5923
# HELP vectorized_alien_total_received_messages Total number of received messages
# TYPE vectorized_alien_total_received_messages counter
vectorized_alien_total_received_messages{shard="0"} 1033
# HELP vectorized_alien_total_sent_messages Total number of sent messages
# TYPE vectorized_alien_total_sent_messages counter
vectorized_alien_total_sent_messages{shard="0"} 13440
# HELP vectorized_application_build Redpanda build information
```

There are several options to modify the behavior of the metrics endpoint,
specifically to emulate different sizes of Redpanda node and Redpanda cluster.

E.g., to approximate a T7 cluster, use the following flags:


```
./metrics-mock.py print --partitions 50000 --topics 10 --shards 31 --nodes 9
```

Each `./metrics-mock.py` instance acts as a single node in a multi-node cluster:
in the above example the mock will act as a single node in a 9 node cluster. This
is important when deciding which metrics appear for resources which are allocated
to a partitcular subset nodes, such as topic partitions. To emulate a differnet
node, pass `--node-id`. If you are emulating a multi-node cluster you should pass
a different node id to each mock, or else they all return the same metrics for
per-topic, per-partition or per-group metrics, which will significantly underestimate
the total metrics count.

---

### Overview (`tools/offline_log_viewer/README.md`)

Offline tool to parse RedPanda disk logs.
Usage: `python viewer.py -h`

---

### Overview (`tools/p/README.md`)

The `p` exectuable script is a containerized drop-in replacement for `p`.

```
tools/p/p compile
```

---

### Overview (`tools/rp_storage_tool/README.md`)

# rp-storage-tool

## Purpose

This low level tool is for offline use by Redpanda Engineering when diagnosing
faults.

This tool is **not** for everyday use on live clusters.

## Quickstart

    # Get a rust toolchain
    curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

    # Compile and run
    cargo run --release -- --backend=<aws|gcp|azure> scan --source=<bucket name>

## Installation

To run the tool outside a built tree, you may simply copy the statically linked binary.

    # Compile
    cargo build --release

    # Grab the binary
    cp target/release/rp-storage-tool /usr/local/bin

## Usage

### Typical usage

The following examples assume you are running on an AWS instance with IAM roles configured
for access to your bucket. If you need to specify static keys, see the "Working with local object stores"
section below.

    # Scan the bucket, store metadata in a file, and report any metadata anomalies
    $ rp-storage-tool --backend aws scan-metadata --source=my-bucket --meta-file my-bucket.json

    # Scan the data within a topic of interest, and report any anomalies.  Use the metadata
    # file from the last step to avoid re-scanning all objects in the bucket.
    $ rp-storage-tool --filter kafka/my-topic/*_* --backend aws scan-data --source=my-bucket --meta-file my-bucket.json

    # Having identified an issue, extract the data from a partition for further analysis offline
    $ mkdir my-data-dump
    $ rp-storage-tool --filter kafka/my-topic/13_63 --backend aws extract --source=my-bucket --meta-file my-bucket.json --sink ./my-data-dump/

### Generic parameters

* The `--backend` argument selects the cloud storage backend. This has a default (AWS) for convenience when working with
  commands that don't use cloud storage, but ordinarily you should be specifying it.
* The `--filter` argument controls which topics & partitions will be examined. When scanning metadata we
  always scan all objects, but will only store+analyze metadata for partitions matching the filter.
* If you are running on a node with authentication already set up (e.g. IAM Roles on AWS), this will Just Work.
  Otherwise you may need to set the appropriate access key/secret environment variables for the cloud platform you are
  connecting to.

### Subcommands

#### `scan-metadata`

This walks the objects in a bucket used by Redpanda Tiered Storage and reports on any inconsistencies.
**Not all issues this tool reports are harmful**, for example `segments_outside_manifest` may contain objects harmlessly
left behind when Redpanda was restarted during an upload.

#### `scan-data`

This walks the data within segments (you probably want to use `--filter` to only scan segments for certain NTPS)
to check that it is readable, and that the metadata in the partition manifest is consistent with the data.

#### `extract`

For NTPs matching the filter, copy all metadata and data objects into a local folder.

## Working with local object stores

This tool uses environment variables for any special configuration of storage backends required for working outside of
real cloud environments.

For example, to use the tool with a bucket called `data` in a minio cluster at `aplite:9000`:

    AWS_ALLOW_HTTP=1 AWS_ACCESS_KEY_ID=minioadmin AWS_SECRET_ACCESS_KEY=minioadmin AWS_REGION=us-east-1 \
      AWS_ENDPOINT=http://aplite:9000 cargo run --release -- --backend aws scan --source data

### Building a portable binary

If you are building on a workstation and copying the binary to a remote machine, and your
workstation doesn't happen to run the same linux distro as the remote machine, it is
useful to build a statically linked binary that will work on any distro.

    # A dependency requires this to build: adjust as needed if not on an RPM distro
    sudo dnf install -y musl-gcc

    # Install the Rust toolchain for musl (statically linked libc)
    rustup target add x86_64-unknown-linux-musl

    # Build with the musl toolchain
    cargo build --release --target=x86_64-unknown-linux-musl

    # Output in target/release/rp-storage-tool

---

### Overview (`tools/type-checking/README.md`)

# Ducktape Type Checking

This directory contains the infrastructure for running PEP-484 type checking on Redpanda's Python test suite (ducktape) using [pyright](https://github.com/microsoft/pyright).

## Overview

The type checking system provides:
- **Gradual type checking adoption** with per-file strictness levels
- **Automated promotion detection** to identify files ready for stricter checking
- **CI integration** to prevent type regressions
- **Docker-based execution** for consistent environments

## Files

### `type-check.py`

The main type checking orchestration script. Provides multiple commands:

- **`check`** - Run type checking on files and report errors
- **`promotion-check`** - Identify files that can be promoted to stricter type checking levels
- **`fruit`** - Show files with fewest errors at each target level (low-hanging fruit for type checking improvements)
- **`ci`** - Run both `check` and `promotion-check` and decorates the results (used in CI)
- **`check-sorted`** - Verify that `type-check-strictness.json` is sorted
- **`pre-commit`** - Run `check` on changed files (intended for use as a pre-commit hook); accepts paths rooted at repo root (`tests/...`) and strips the prefix internally

### `type-check-strictness.json`

Configuration file that maps Python files to their current type checking strictness level. Files not listed default to `strict` level.

**Strictness levels** (from most to least strict):
- `strict` - Full type checking (default for unlisted files)
- `standard` - Standard type checking
- `basic` - Basic type checking
- `off` - Minimal checking (syntax, imports, etc.)
- `skip` - No type checking at all

### `Dockerfile`
Multi-stage Docker image for running type checking in a consistent environment. Installs Python dependencies and pyright.

### `type-check.sh`
Shell script that builds the Docker image and runs type checking inside a container.

### `requirements.txt`
Python dependencies for type checking (primarily pyright).

## Usage

### Local Development

#### Using task

There is a type-check `task` which is a very thin wrapper around the scripts: this is probably the most convenient way to call it as it works
anywhere in the repository. Call it with arguments for the script after `--`, like so:

```
task rp:type-check -- --help
```

#### Using type-check.sh

You can also call the script as `tools/type-checking/type-check.sh --help`. This is essentially the same as the `task` invocation above. This runs the underlying python script inside a docker container, so it should "just work".

#### Using type-check.py

Locally you should also able to run the .py script directly, like `tools/type-checking/type-check.py --help`  if you have run [local_venv.sh](../../tests/local_venv.sh).

### Examples

These examples all the `task` invocation method but work with any of the methods above.

```bash
# Run type checking on all files
task rp:type-check -- check

# Check for files ready to be promoted to stricter levels
task rp:type-check -- promotion-check

# Find low-hanging fruit - files with fewest errors at each target level
task rp:type-check -- fruit

# Show top 25 fruit candidates
task rp:type-check -- fruit --top 25

# Show detailed error information with verbose output
task rp:type-check -- fruit --verbose 2

# Automatically promote eligible files
task rp:type-check -- promotion-check --update

# Run CI checks (both check and promotion-check)
task rp:type-check -- ci

# Check specific files (paths/globs are positional, relative to tests/)
task rp:type-check -- check rptest/tests/my_test.py

# Force all files to be checked at a specific level
task rp:type-check -- check --force-level strict
```

## Strictness Level Management

In order to have type checking useful immediately, without needing to fix 1000s of existing type errors, the system uses a progressive approach to type checking adoption:

1. **New files** default to `strict` type checking
2. **Existing files** are assigned to the strictest level they currently pass
3. **Promotion detection** automatically identifies when files can move to stricter levels
4. **Automated updates** can promote files with `--update` flag

### Adding Type Hints to Files

When adding type hints to improve a file:

1. Run `tools/type-checking/type-check.py promotion-check --update`
2. This will automatically move the file to the strictest level it now passes
3. Commit both your type hint changes and the updated `type-check-strictness.json`

### Finding Low-Hanging Fruit

The `fruit` command helps identify files that are closest to being promotable to stricter type checking levels by showing files with the fewest errors at each target level:

```bash
# Show top 10 files with fewest errors per target level
task rp:type-check -- fruit

# Show top 20 files with fewest errors per target level
task rp:type-check -- fruit --top 20

# Show detailed error information with verbose output
task rp:type-check -- fruit --verbose 2
```

This is useful for:
- **Prioritizing type checking improvements** - start with files that need minimal fixes
- **Setting realistic goals** - see how many errors need to be fixed for each promotion
- **Understanding error distribution** - identify patterns in type checking issues

The command groups files by their potential promotion target (off→basic, basic→standard, etc.) and sorts them by error count, making it easy to find files that are closest to passing at the next strictness level.

### Failed Promotion Check

You may be here because CI failed with a "promotion check" error. This means that the strictness level for some files are set too low, i.e., your change has improved the state of the Python code base and as a reward you can promote some files to higher strictness levels. Pat yourself on the back, then use the following command to automatically update the strictness file:

```bash
# Automatically update the configuration
task rp:type-check -- promotion-check --update

# Commit the updated type-check-strictness.json
git add tools/type-checking/type-check-strictness.json
git commit -m "chore: update type checking strictness"
```

## Command Line Options

- `--tests-root PATH` - Path to tests directory (auto-detected by default)
- `--config PATH` - Path to pyrightconfig.json
- `input_files` - Positional paths/globs of files to check, relative to `tests/` (default: `rptest/**/*.py`)
- `--force-level LEVEL` - Override strictness levels and check all files at specified level
- `--no-venv` - Don't use virtual environment when running pyright
- `--verbose [LEVEL]` - Enable verbose output (0=off, 1=verbose, 2=very verbose)
- `--update` - Update type-check-strictness.json in-place (for promotion-check)
- `--top N` - Number of files to show per target level in fruit command (default: 10)

## CI Integration

The type checking runs automatically on pull requests via GitHub Actions (`.github/workflows/python-type-check.yml`). The CI:

1. Builds the Docker image from `Dockerfile`
2. Runs `type-check.py ci` which executes both type checking and promotion detection
3. Fails if there are type errors or files that should be promoted

This check tries to be as fast as possible, and usually runs in about 3 minutes if the docker image build is fully cached (usually the case), or a few minutes more if not.

## Architecture

The system is designed around Microsoft's pyright type checker with these key features:

- **Segmented checking**: Files are grouped by strictness level and checked separately
- **Temporary configs**: Generates temporary pyrightconfig.json files for each strictness level
- **Batch processing**: Efficiently checks multiple files at once when possible
- **Virtual environment integration**: Uses `.venv` if available for consistent Python environment

## Adding New Features

The `type-check.py` script is modular and can be extended with new commands by:

1. Adding the command name to the `CMDS` list
2. Adding documentation to `COMMAND_DOC`
3. Implementing a method on the `TypeCheck` class with the command name (with hyphens converted to underscores)

The script follows the pattern where each command method returns `True` for success, `False` for failure, or `None` for success (which is treated as `True`).

---

### Overview (`bazel/clang_tidy/plugins/README.md`)

# Redpanda clang-tidy plugins

This directory contains sources and bazel targets for custom out-of-tree clang-tidy checks.
It contains a single `redpanda_cc_binary` target, `plugins.so`, which contains a single clang-tidy
module, `redpanda-module`.

## Building

```sh
$ bazel build //bazel/clang_tidy/plugins:plugins.so
```

## Adding a new check

1. Write your check in a separate file or files. e.g. `plugins/redpanda_noop_check.h`.
2. Add your sources directly to `plugins.so.srcs` including any new dependencies.
3. Register the check by name in `plugins/redpanda_tidy_module.h`.
   - The name should start with "redpanda-".

Checks introduced in this way are automatically loaded in `bazel build --config=clang-tidy //...`.

If you want to try your check on some source file in isolation, you can invoke the tool directly:

```sh
$ bazel build //bazel/clang_tidy/plugins:plugins.so
$ ./tools/clang-tidy --checks=-*,redpanda-my-fancy-check -list-checks
$ ./tools/clang-tidy --checks=-*,redpanda-my-fancy-check /path/to/some/foo.cc
```

---

### Overview (`proto/redpanda/datalake/README.md`)

# Redpanda Datalake Protocol Buffers

This package contains definitions of types for datalake integration
(i.e. Iceberg) that are otherwise impossible to express with built-in
types or existing protobuf well known types.

Note: These definitions should be usable by end-users as-is! They shouldn't
contain Redpanda specific implementation details (like customization of the
C++ namespace).

---

### Overview (`src/transform-sdk/cpp/README.md`)

# Experimental Redpanda Data Transforms C++ SDK

This directory contains an experimental C++ SDK for Redpanda Data Transforms.
C++23 is required for both language features and standard library support.

When writing C++ in WebAssembly, note that certain language features are not supported.
Noteably C++ exceptions, `setjmp`/`longjmp`, threads and networking is not supported, WebAssembly and WASI are quickly evolving, so some of these features may be supported in the future.

## Usage 

A specialized clang toolchain is needed for WebAssembly + WASI compilation. The easiest way to
do this is using [zig cc][zig] or [WASI SDK][wasi-sdk] via docker.

First create the identity transform that mirrors records to the output topic from the input topic.

Create your `transform.cc` file that does this:

```cpp
#include <redpanda/transform_sdk.h>

int main() {
    redpanda::on_record_written([](redpanda::write_event event, redpanda::record_writer* writer) {
      return writer->write(event.record);
    });
}
```

### CMake

Using CMake create a `CMakeLists.txt` with the initial project boilerplate, which pulls down our C++ SDK and builds the transform.

```cmake
cmake_minimum_required(VERSION 3.22)

project(
  MyTransform
  VERSION 0.1
  LANGUAGES CXX)

set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
set(CMAKE_CXX_STANDARD 23)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_CXX_EXTENSIONS OFF)

add_compile_options(-Wall)

set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-exceptions")

include(FetchContent)
set(FETCHCONTENT_QUIET FALSE)

FetchContent_Declare(
  redpanda-transform-sdk
  URL                    https://github.com/redpanda-data/redpanda/archive/refs/heads/dev.tar.gz
  SOURCE_SUBDIR          src/transform-sdk/cpp
)

FetchContent_MakeAvailable(redpanda-transform-sdk)

add_executable(
  my_transform
  transform.cc
)
target_link_libraries(
  my_transform
  Redpanda::transform_sdk
)
```

Once this file is created generate the build and build the `my_transform` Wasm binary via:

```shell
docker run -v `pwd`:/src -w /src ghcr.io/webassembly/wasi-sdk /bin/bash -c 'cmake -Bbuild && cmake --build build'
```

### Standalone

It can be compiled to WebAssembly using `zig`, the result will be a WebAssembly binary named `a.out` in the current directory.

```shell
zig c++ --target=wasm32-wasi -std=c++23 -fno-exceptions -flto -O3 -Iinclude transform.cc src/transform_sdk.cc
```

Alternatively using docker and WASI SDK, the same program can be compiled using the following:

```shell
docker run -v `pwd`:/src -w /src ghcr.io/webassembly/wasi-sdk /bin/bash -c '$CXX $CXX_FLAGS -std=c++23 -fno-exceptions -O3 -flto -Iinclude transform.cc src/transform_sdk.cc'
```

## Development

CMake can be used for local development. There is a small test suite embeded that can be compiled via `-DREDPANDA_TRANSFORM_SDK_ENABLE_TESTING`. We keep the code to a single header and source file so it is easy to drop into other build systems.

[zig]: https://andrewkelley.me/post/zig-cc-powerful-drop-in-replacement-gcc-clang.html
[wasi-sdk]: https://github.com/WebAssembly/wasi-sdk

---

### Overview (`src/transform-sdk/js/README.md`)

# Redpanda Data Transform JS SDK

This holds the JS SDK code for Redpanda's Data Transforms. This SDK is implemented by
compiling the QuickJS JavaScript VM into WebAssembly and running it within our C++ SDK.

To use this SDK manually here are the steps:

```shell
# Compile the C++
docker run -v `pwd`/..:/src -w /src/js ghcr.io/webassembly/wasi-sdk \
  /bin/bash -c 'apt update && apt install -y git && cmake --preset release-static && cmake --build --preset release-static -- redpanda_js_transform -j32'
# Convert the JavaScript file into a WebAssembly text file
cat user_code.js | ./generate_js_provider.py > js_user_code.wat
# Download wasm merge from Binaryen, and use it to merge the wasm files together, resolving the imports in C++ for the source code.
wasm-merge build/release-static/redpanda_js_transform js_vm ./js_user_code.wat redpanda_js_provider -mvp --enable-simd --enable-bulk-memory --enable-multimemory -o js_transform.wasm
# Deploy away!
rpk transform deploy --file=js_transform.wasm --name=my-transform -i foo -o bar
```

There is still a lot of work todo to provide a full experience of NodeJS compatibility, but this is a start.

---

### Overview (`src/transform-sdk/rust/README.md`)

# Redpanda Data Transforms Rust SDK

This crate contains the SDK for Redpanda's inline Data Transforms, based on WebAssembly.

Data transforms let you run common data streaming tasks, like filtering, scrubbing, and transcoding, within Redpanda. For example, you may have consumers that require you to redact credit card numbers or convert JSON to Avro.

Data transforms use a WebAssembly (Wasm) engine inside a Redpanda broker. A Wasm function acts on a single record in an input topic. You can develop and manage data transforms with [`rpk transform`][rpk-transform] commands.

See [`examples/`][examples] for sample usages.

[rpk-transform]: https://docs.redpanda.com/beta/develop/data-transforms/run-transforms/
[examples]: https://github.com/redpanda-data/redpanda/tree/dev/src/transform-sdk/rust/examples

---

### Overview (`src/v/base/README.md`)

The base library is a foundational library. The rough guidelines for determining
if something belongs in the base library are:

1. It would be reasonable if every target depended on it (e.g. vlog.h).
2. It contains no dependencies on other Redpanda targets (e.g. utils).
3. Its external dependencies may also be considered foundational (e.g. seastar).

---

### Overview (`src/v/cloud_io/README.md`)

# Cloud IO

Encapsulates IO related to remote storage. Callers should interact with this
code via generic interfaces, like streams, files, and buffers, building logical
abstractions on top (e.g. partition manifests and friends in `/cloud_storage`)
and hiding the details of per-backend client connections (e.g. in
`/cloud_storage_clients`).

### Why does this exist?

As new cloud storage applications are developed, lumping everything into
`/cloud_storage` won't be sustainable, as this would encourage monolithic
classes that are more easily maintained split up.

### What code should go here?

A high level goal is to move re-usable code into `/cloud_io` and leave
abstractions specific to tiered storage in `/cloud_storage`. Caches of bytes or
cached files belong here; caches for segments or batches do not.

---

### Overview (`src/v/cloud_topics/README.md`)

```mermaid
sequenceDiagram
    box Kafka Partition Leader (Client Perspective)
    participant Kafka
    participant Frontend
    participant Batcher
    end
    box
    participant Object Storage
    end
    box CT Partition Leader
    participant Partition Leader
    end
    participant Frontend as CT Frontend
    Kafka->>Frontend: Produce Batch
    Frontend->>Batcher: Batch Payload
    Batcher->>Object Storage: L0 Upload
    Object Storage-->>Frontend: Upload Ack
    Frontend->>Partition Leader: Replicate placeholder
    Partition Leader-->>Kafka: Produce Ack
```

### Missing bits

* In the write path we should populate batch cache with `raft_data` batches
  instead of `ctp_placeholder` batches. Currently, the `ctp_placeholder` batches
  are replicated and so they're added to the batch cache. If we want low to
  achieve low latency for tailing consumers we need to make sure that they're
  reading from the batch cache.
* Reader caching. Currently, the code just consumes from the partition and then
  creates a `placeholder_extent` for every `ctp_placeholder` batch. Then the
  extent gets materialized (which involves expensive I/O). The data pulled by
  the `placeholder_extent` contains information that belongs to different NTPs
  but the reader only uses one that belongs to the NTP from which it's reading
  from. We need to cache L0 objects in memory so the data will be reused.
* Resource usage limitations. There should be a mechanism to limit memory use,
  inbound and outbound network and disk bandwidth. This will probably look like
  a per-shard token bucket for every resource. We should also limit retries in
  case of errors in a centralized way (circuit breaker or token bucket).

---

### Overview (`src/v/crypto/README.md`)

# Redpanda Crypto Library

The Redpanda crypto library provides a convenient API for any subsystem that
desires to use a cryptographic function.  It will abstract away the actual
implementation (OpenSSL).

This library provides the following implementations:

* RSA PKCSv1.5 signature verification
* HMAC
* MD5, SHA256 and SHA512 digest generation
* Cryptographically secure DRBG
* Base64 decoding

## Cryptographic Secure Digest

This library currently supports the following digest types:

* MD5
* SHA256
* SHA512

There are two ways of performing a digest: one-shot and multi-part.

For one-shot, you can do the following:

```c++
#include "crypto/crypto.h"

bytes msg = {...};
auto digest = crypto::digest(crypto::digest_type::SHA256, msg);
```

For multi-part operation, you can do the following:

```c++
#include "crypto/crypto.h"

bytes msg = {...};
crypto::digest_ctx ctx(crypto::digest_type::SHA256);
ctx.update(msg);
auto digest = std::move(ctx).final();
```

## HMAC

Like digests, this library can perform HMAC two ways: one-shot and multi-part.

For one-shot:

```c++
#include "crypto/crypto.h"

bytes key = {...};
bytes msg = {...};

auto sig = crypto::hmac(crypto::digest_type::SHA256, key, msg);
```

For multi-part:

```c++
#include "crypto/crypto.h"

bytes key = {...};
bytes msg = {...};

crypto::hmac_ctx ctx(crypto::digest_type::SHA256, key);
ctx.update(msg);
auto sig = std::move(ctx).final();
```

## Asymmetric Key Handling

Currently, this library only supports RSA keys.  There are two ways of loading a
key from a buffer.

First, if the key is held within a buffer in PKCS8 format, you can use
`crypto::load_key`:

```c++
#include "crypto/crypto.h"

bytes rsa_priv_key {...};

auto key = crypto::key::load_key(
  rsa_priv_key,
  crypto::format_type::PEM,
  crypto::is_private_key_t::yes);
```

The above function can determine the type of key held in the buffer but the
caller is responsible for indicating the format the key is in (PEM or DER) and
whether or not it's the public half of the key or the private key.

The other way of loading a key is by its parts.  Currently only RSA public key
loading is available:

```c++
#include "crypto/crypto.h"
bytes modulus {...};
bytes public_exponent {...};

auto key = crypto::key::load_rsa_public_key(modulus, public_exponent);
```

## Signature Verification

Performing signature verification can be done either one-shot or multi part.

One-shot:

```c++
#include "crypto/crypto.h"

auto key = {...};
bytes msg {...};
bytes sig {...};

auto sig_verified = crypto::verify_signature(
  crypto::digest_type::SHA256,
  key,
  msg,
  sig
);
```

For multi-part:

```c++
#include "crypto/crypto.h"

auto key = {...};
bytes msg {...};
bytes sig {...};

crypto::verify_ctx ctx(crypto::digest_type::SHA256, key);
ctx.update(msg);
bool verified = std::move(ctx).final(sig);
```

## RNG Generation

To get cryptographically secure random data, please use
`crypto::generate_random`.  There are two functions you can use.  One will
place random data into the provided buffer and another will return an allocated
buffer with random data.  Both functions also have a flag to indicate if the caller
wants to use the 'private' DRBG.  For more information on this please refer to
OpenSSL's [documentation](https://www.openssl.org/docs/man3.0/man3/RAND_priv_bytes.html)
on this.

---

### Overview (`src/v/features/README.md`)

# features

## Why does this file get its own top-level library?

The feature table may be accessed from many different parts
of redpanda, and is instantiated in application.cc very early
in application startup.

---

### Overview (`src/v/metrics/README.md`)

# Adding metrics

Together with logs metrics are one of our core tools for observability and
debugability. Many incidents, bugs or performance issues can purely be solved
based on metrics. Hence adding new metrics to further improve our insight into
Redpanda is highly encouraged.

However we do already have a huge amount of metrics. Depending on certain
parameters such as partition count a prometheus scrape might have millions of
lines for certain clusters. This causes problems at scrape and ingestion time
which can ultimately lead to a total loss of metrics. 

Hence we need to be mindful of how many metrics we add. Below is a guide for
things to watch out for when adding new metrics. 

## Metrics Aggregation

One thing to keep in mind when adding new metrics is “metrics aggregation”. In
our context this refers to a seastar feature where on metric creation one can
specify “aggregation labels”. These are labels which will be “aggregated away”.
For example if you had N series of a metrics family all of which have a
different value for a certain label and you specified that label as an
aggregation label then the resulting prometheus output would have a single
series for that metric family with the label removed and the value being the sum
of all the aggregated series. 

Note above I put quotes around many instances of “aggregation” which is because
aggregation can mean the opposite in the prometheus sense. I.e.: if you do
something like a sum by (foo, bar) then actually all but foo and bar will be
“aggregated away”.

To solve the aforementioned problem of overloading our metrics system we employ
metric label aggregation to reduce the count of metrics series we produce.

In a few cases some metrics are defined with static aggregation labels. This
rarely makes sense as one could just avoid adding the extra labels in this case.
However the one exception is the shard label which gets automatically attached
to each metric series depending on which shard it is coming from. In rare cases
it’s actually uninteresting on which shard something happened so we might always
unconditionally aggregate the shard label away.

In the majority of cases we determine the aggregation labels dynamically based
on the aggregate_metrics config flag. If turned off we usually don’t aggregate
any labels and expose the full set of labels while if it’s turned on we usually
aggregate a subset of labels away. Common ones are the shard, partition or topic
label. Basically any label that can result in high cardinality.

Unless you have a good reason not to you’ll generally want to aggregate any
label that can scale unbounded (with for example partitions or topics). As of
today we also always aggregate the shard label in this case. Though this is
something we might want to change in the future.

Metrics aggregation is on by default in Redpanda Cloud. As of 23.3+ the config flag can
be changed dynamically at runtime without the need for a restart. Hence it’s
easy to get access to all the labels when needed. 

Note that when talking about metrics overhead we mostly talk about downstream
systems of redpanda (i.e.: prometheus itself, grafana etc.) metrics collection
itself has a measurable overhead on redpanda itself. Metrics aggregation
amplifies this as it kind of trades CPU in redpanda for less IO/CPU further
downstream. Hence metrics aggregation is not a wildcard to adding a massive
amount of labels that are later being aggregated away. 

## Adding new metrics

### Counters / Gauges

Adding counters and gauges is generally always fine as those just add a single
new series per shard. Having a static set of extra labels is also fine (e.g.:
the handler type of the kafka API) as those can’t grow unbounded.

Think twice whether labels that can scale dynamically are really needed, i.e.:
topic/partition labels. Always aggregate those when aggregation is turned on.

### Histograms

More care needs to be taken when adding histograms. A single histogram adds more
than 26 metrics series on internal metrics. Hence a histogram has more than 10x
the overhead of a counter or gauge.

This means that any additional label should be considered carefully. Adding
dynamic labels such as partition or topic labels should be avoided altogether.
Keep static labels such as the handler type for the core/really important
histograms. In that example consider whether a separate histogram for
produce/consume/all-other-handlers would be enough. 

### Topics vs. partitions

As mentioned previously the two label types that really cause problems are
partition and topic labels. These can scale dynamically/unbounded and massively
increase the metrics cardinality.

Note that one might think that the topic label should add a lot less series than
the partition label. While this is true most of the time it is unfortunately not
always the case. Some customers might create many many topics (think topic per
“client” or something) with relatively little partitions.

Hence you should think of topic having the same cardinality as partition. 

## Public metrics

In Redpanda we have two different metric endpoints /metrics and /public_metrics.
The former is sometimes referred to as the "internal" endpoint. In Redpanda
Cloud only the public endpoint is exposed.

Public metrics are intended to be low-volume and high-value. They should provide
essential observability without overwhelming downstream systems. The public
endpoint is typically up to 100x smaller than the internal one.

Because of this requirement the rules are more strict in terms of how many
metrics series to add. Aggregation is always enabled on the public endpoint
regardless of the aggregate_metrics config flag.

When adding metrics to the public endpoint keep these in mind:

- Aggregate most labels to avoid any form of scaling with shard, partition, etc.

- Consider whether the metric fits the public metrics criteria. For example, an
  internal buffer ratio might be helpful for debugging but does not qualify as a
  public metric. Most metrics fall into this category and belong only on the
  internal endpoint.

---

### Overview (`src/v/migrations/README.md`)

# Migrators

## What is this?

A migrator class listens for the status of a particular feature
in the feature table, and responds to the `preparing` state by
taking some necessary data migration action.  This might mean
rewriting an old data format to a new format, or tweaking a 
configuration from a legacy system to fit a new configuration
scheme.

The name comes from the analogy with database migrations.

## Why are they here?

Migrators are segregated in their own directory because they
are intrinsically temporary code: after a safe number of releases
have passed, we may retire them.

This enables adding & removing migrators without disrupting the
areas of code they migrate data on behalf of.

---

### Overview (`src/v/random/README.md`)

The `random` library is a foundational library that provides access to random
data generation utilities for primitive types such as integers and characters.

## Main Entry Points

**For non-cryptographically secure random numbers**, use `random_generators`:
- Fast, high-quality pseudorandom number generation using PCG64
- Suitable for simulations, testing, sampling, and general-purpose randomness
- Configurable seeding modes for deterministic testing vs random production behavior
- Thread-safe global instance available via `random_generators::global()`
- Provides common distributions like `get_int()` for uniform integer generation
- For more specialized distributions (normal, exponential, etc.), use standard
  library `<random>` distributions or Abseil distributions with the underlying
  engine accessible via `rng::engine()`
- For details on seeding behavior see [random values in tests](https://redpandadata.atlassian.net/wiki/spaces/CORE/pages/1406271495/Random+values+in+tests)

**For cryptographically secure random numbers**, use `secure_random`:
- Cryptographically secure random number generation
- Suitable for security-sensitive applications like key generation, nonces, etc.
- Uses system entropy sources

## Usage Guidelines

Modules that want to provide random generation for higher-order types should
link against the `random` library and provide specialized generators using
the appropriate underlying primitives based on their security requirements.

---

### Overview (`src/v/schema/README.md`)

# Schema Library

This library contains functionality to interact with the schemas in the Schema Registry, hiding a bulk of the internal details found within `pandaproxy::schema_registry`. Developers should consider using this instead of depending directly on `pandaproxy` if their code only needs application-facing primitives like schemas, schema IDs, subjects, etc., and not e.g. specific HTTP endpoints or the `_schemas` topic.

---

### Overview (`src/v/transform/README.md`)

# Data Transforms Library

This library contains the control and data plane that powers Data Transforms. Data Transforms as an entire component is divided into 4 major parts:

1. Wasm virtual machine (Wasmtime)
1. Wasm integration for data transforms (src/v/wasm/ module)
1. Data transforms distributed system layer (this module)
1. Data transforms SDKs that run inside the VM (src/transform-sdk)

At a high level this library defines the architecture for how data flows through transforms. There are two major components: transform_manager and transform_processor representing the control plane and the data plane respectively.

The two planes are loosely coupled in an event emitting pattern. The control plane is a single fiber loop that responds to events such as raft leadership notifications for partitions, and data plane errors. The control plane loop then acts accordingly by scheduling, starting and shutting down corresponding transform_processors. The data plane is 3 fibers to read, transform and write data, all connected via buffers (defined in `transform_queue.h`).

Other parts of the subsystem are the commit batcher, which recieves requests to commit processed offsets for individual transform processors. The batcher periodically (according to cluster configuration) commits all the offsets on that core at once, so that the number of commits scales with the number of cores, not the number of partitions.

Most IO in this subsystem is abstracted into interfaces so that they can be swapped out and unit tested. The actual implementation and stitching up of the production components is all done in `api.cc`. The underlying operations are mostly implemented in the transform/rpc subsystem, which defines the internal data path for transforms (to circumvent the kafka API) over our internal RPC protocol.

---

### Overview (`src/v/wasm/README.md`)

# WebAssembly Engine Library

This library contains the bulk of the functionality related to WebAssembly Virtual Machines and their integration into Redpanda.
Currently, this library's focus is in support of Data Transforms, so the APIs are focused around that. If additional functionality
for Wasm was introduced to Redpanda in the future, the `wasm::engine` abstraction would need to split, as there would be an engine
in support of each plugin functionality needed.

The major abstractions of this package are listed below. Note that it's important that these abstractions are solid - we take advantage
of the abstractions in numerous places to provide functionality like caching/reuse of engines on a core, dummy engines for testing or even
engines that exist on different computers.

#### **Runtime**

The main entry point into this library, it holds the configuration for the entire subsystem and has the ability to compile a Wasm binary
into a _Factory_. There is only a single logical runtime for a process and access to that runtime is normally coordinated on shard 0.

#### **Factory**

A factory is a tricky object. A factory represents a compiled artifact that can cheaply be made into engines for a given core. Since 
compilation and the resulting artifacts are very expensive, we only create one per process and share them around cores (use 
`seastar::foreign_ptr`). But, whichever core makes an engine using this factory, the engine must only be used on that core.
This can be a little mind boggling to work through, but thankfully most users of this API can just ask for an engine on their 
core, and this logic can be a single isolated function.

#### **Engine**

Engine is the workhorse of the data plane for Data Transforms. It's the part that actually takes batches and writes them into the Wasm
runtime then reads the results back out. In principle it's quite simple, but in reality it's a bit of a dance to do this in a high 
performance manner. There are various modules that are stored within an engine and exposed to the guest code running inside the VM.
See the FFI section for more information on how to extend and add new functionality.

## Wasmtime

Our WebAssembly VM is built using [Wasmtime][wasmtime]'s [C-API][c-api]. Wasmtime itself is a very high performance Wasm VM, but is still
lightweight and is customizable in all the ways we require. For example, by default Wasmtime `mmap`'s memory for stacks and linear memory
inside of the VM at runtime, but also allows plugging in custom allocators (which we use as Seastar has strict opinions about memory). To learn
more about the allocators see `allocator.h`.

We use Wasmtime's ability to run the guest code on another stack extensively. In wasmtime's documentation you'll see this referenced as "async" APIs.
Wasmtime will execute the guest using an explicit stack, and switch back to the original caller's stack to support suspending the VM. This is how host
functions can return futures. We tell the VM to switch back and then once we've switched back we tell Seastar's scheduler to re-schedule running the VM
after the host function's future has resolved - the details here are in `invoke_async_host_fn` and `wasmtime_engine::_pending_host_function`.

#### Caveats

Wasmtime uses Unix signals extensively to implement specific behaviors (ie aborts in the VM), which has in the past caused issues with Seastar. 
Seastar by default blocks (most) signals for all shards except shard0, and has different behaviors in debug mode when ASAN is enabled. To combat this
we unblock all signals that Wasmtime needs on startup for all cores when Wasm is enabled. If Wasmtime changes the signals they use we will want to
update our list of signals we force Seastar to unblock. Lastly, there are some complications with seastar::thread and it usage of `swapcontext` that
can preserve signals and reblock them at times. Thankfully since we don't use seastar::thread much in production this isn't an issue in practice.

## Foreign Function Interface (FFI)

WebAssembly on it's own is completely isolated/sandboxed and has no ability to interact with the host or outside. We have a small custom framework
we've written to make it easy (-ish) to expose C++ classes into the VM. The general conversion code between C++ methods and Wasm memory is in `ffi.h`,
but the actual registration code and hooking that up is in `wasmtime.cc` - in particular the `host_function::reg` function, which is generally wrapped
in a `REG_HOST_FN` macro for the modules we support.

### Current Modules

We currently have the following list of modules exposed to guest code running inside the VM. How one would call these modules from a given language
can be a little mysterious, but we currently (at time of writing) have examples checked into the tree for C/C++, Golang and Rust.

#### transform_module

The transform module is the heart of transforming models and has a tight integration with the `wasmtime_engine` class. This is because we always have
the engine "active" so to speak. All of our SDKs at their core is a loop like the following:

```python
def main():
    check_abi()
    while True:
        n_records = read_batch_header()
        for _ in range(n_records):
            r = read_record()
            for out in transform(r):
                write_record(r)
```

When there are no batches left `read_batch_header` suspends the VM until another record comes along. Most of the complexity of the VM lifetime is due to this
approach where the VM polls the host for records (as opposed to the broker calling into the VM for each record to transform). On similar note - the reason we
use a per record interface to the VM instead of a per batch interface is for a couple of reasons. One, we can easily set limits like timeouts per record and they
become more intuitive for our users (who think per record). Two, record sizes are generally less variable than batch sizes, so it becomes easier to size functions
for memory requirements.

#### schema_registry_module

In order to access schema registry, we expose a custom module for using a few key schema registry APIs. We wrap the schema registry subsystem in a custom
class because it's easier to use and we also are able to swap out that class for testing. See `schema_registry.h` for the details.

#### wasi_module

WASI is a standard for WebAssembly interacting with POSIX-like host APIs such as filesystems, clocks, environment variables, etc. We provide a custom
implementation of these APIs because most upstream toolchains/languages assume at least a POSIX environment, and it's a more familar programming
experience for our users. We just stub out most of the APIs, but some of them (like environment variable support) is fully there.

[wasmtime]: wasmtime.dev
[c-api]: https://docs.wasmtime.dev/c-api/

---

### Overview (`tests/go/go-kafka-serde/README.md`)

# Golang Kafka Serde Client

This directory contains a client that can be used to validate the operation of Redpanda's Schema Registry via
Confluent's Golang client.

This application will create either a simple AVRO or PROTOBUF schema, serialize a number of messages

## Building

### Pre-requisites

* Protobuf Compiler
* Golang compiler

### Build Instructions

It is important to set `$GOPATH` either within your environment or on the command line when you run `make`:

```shell
GOPATH=$HOME/go make clean all
```

## Usage

```shell
❯ ./go-kafka-serde -help
Usage of ./go-kafka-serde:
  -brokers string
        comma delimited list of brokers (default "localhost:9092")
  -consumer-group string
        Consumer group to use (default "db5b01db-7f48-4d7d-8ba6-eb512788341e")
  -count int
        Number of messages to produce and consume (default 1)
  -debug
        Enable verbose logging
  -protocol string
        Protocol to use.  Must be AVRO or PROTOBUF (default "AVRO")
  -schema-registry string
        URL of schema registry (default "http://127.0.0.1:8081")
  -security string
        Security settings
  -topic string
        topic to produce/consume from (default "3aa75595-1222-4daa-b407-afe43c8496d5")
```

---

### Overview (`tests/go/kgo-verifier/README.md`)

# kgo-verifier & kgo-repeater

These are Kafka traffic generators using franz-go for driving significant
throughput on a Redpanda cluster.

## kgo-repeater

### Purpose

Scalable (across multiple nodes) population of producers and consumers.  Each
worker is a member of a consumer group, and echos consumed messages back to
the cluster as a producer.  This scheme provides straightforward scaling,
as starting additional workers automatically re-distributes production
work alongside the consumption work.

kgo-repeater does not validate the messages it receives, beyond that they
should be syntactically valid and contain the fields it expects: for validation
that messages arrive in the right order and are not lost, use kgo-verifier.

kgo-repeater also provides basic latency measurement, although it is not
engineered to be a rigorous benchmark.

### Examples

#### Single standalone process

    kgo-repeater --brokers 127.0.0.1:9092 -group mygroup -workers 4 -topic mytopic

That will run 4 clients, each consuming and producing.  A
default 4MB of data will be injected into the topic on
startup, and then repeated back through the clients
until the process is signalled to stop.

#### Group of processes

While you can run multiple independent kgo-repeater processes using different
consumer groups, to have multiple processes share the same consumer group (
thereby share work neatly across the clients and exercise larger consumer groups),
there is an HTTP interface to remotely control a group of processes.

The main reason for this remote control is to coordinate startup: ensuring that
the data is not injected into the topic under test until the workers have all
come up and joined the consumer group (otherwise data might be lost and we can
end up driving less data through the system than we intended to)

For example, start two processes with the `-remote` parameter:
    kgo-repeater --brokers 127.0.0.1:9092 -group mygroup -workers 4 -topic mytopic -remote -remote-port 7884
    kgo-repeater --brokers 127.0.0.1:9092 -group mygroup -workers 4 -topic mytopic -remote -remote-port 7885

These processes will start their consumers, but not produce anything until
requested to via an HTTP operation.  To start the producers:

    curl -X PUT localhost:7884/activate
    curl -X PUT localhost:7885/activate

You can then query the progress of the processes:

    # The result is a vector of dicts, one per worker
    # Latencies are in microseconds.
    curl -X get localhost:7884/status
    [
    {
        "produced": 6719,
        "consumed": 6489,
        "enqueued": 25,
        "errors": 0,
        "latency": {
        "ack": {
            "p50": 5128.5,
            "p90": 6379.5,
            "p99": 12562.5
        },


## kgo-verifier

### Purpose

This is a test utility for validating Redpanda data integrity under
various produce/consume patterns, especially random reads that stress
tiered storage.

Stress test redpanda with concurrent random reads, a particularly important
case for validating tiered storage (shadow indexing) where random reads
tend to lead to lots of cache promotion/demotion.

The tool is meant to be chaos-tolerant, i.e. it should not drop out when brokers become
unavailable or cannot respond to requests, and it should continue to be able to validate
its own output if that output was written in bad conditions (e.g. with producer retries).

Additionally verify content of reads, to check for offset translation issues:
the key of each message includes the offset where the producer expects it to land.
The producer keeps track of which messages were successfully committed at the expected
offset, and writes it out to a file.  Consumers then consult this file while reading,
to check whether a particular offset is expected to contain a valid message (i.e. key
matches offset) or not.

### Usage

- Use of TLS is allowed (through `--enable-tls`) with the caveat that the certificate
  must be signed by a known/trusted CA (so no self-signed or self generated CAs)

#### 1. Quick produce+consume smoke test: produce and then consume in the same process

    kgo-verifier --brokers $BROKERS --username $SASL_USER --password $SASL_PASSWORD --topic $TOPIC --msg_size 128000 --produce_msgs 10000 --rand_read_msgs 10 --seq_read=1


#### 2. A long running producer

Run exactly one of these at a time, it writes out
a valid_offsets_{topic}.json file, so multiple concurrent producers would 
interfere with one another

    kgo-verifier --brokers $BROKERS --username $SASL_USER --password $SASL_PASSWORD --topic $TOPIC --msg_size 128000 --produce_msgs 10000 --rand_read_msgs 0 --seq_read=0

#### 3. A sequential consumer.

Run one of these inside a while loop to continuously stream
the whole content of the topic.

    kgo-verifier --brokers $BROKERS --username $SASL_USER --password $SASL_PASSWORD --topic $TOPIC --msg_size 128000 --produce_msgs 0 --rand_read_msgs 0 --seq_read=1 


#### 4. A parallel random consumer
The --parallel flag says how many read fibers to run concurently

    kgo-verifier --brokers $BROKERS --username $SASL_USER --password $SASL_PASSWORD --topic $TOPIC --msg_size 128000 --produce_msgs 0 --rand_read_msgs 10 --seq_read=0 --parallel 4

#### 5. A *very* parallel random consumer
aims to emit so many concurrent reads
that the shadow index cache may violate its size bounds (e.g. do 64 concurrent
reads of 1GB segments, when the cache size limit is only 50GB).
Keep rand_read_msgs at 1 to constrain memory usage.

    kgo-verifier --brokers $BROKERS --username $SASL_USER --password $SASL_PASSWORD --topic $TOPIC --msg_size 128000 --produce_msgs 0 --rand_read_msgs 1 --seq_read=0 --parallel 64

```

---

### Overview (`tests/go/transform-verifier/README.md`)

# Transform Verifier

A small golang program that verifies that data transforms work as expected. 

Most tests for Redpanda today don't inspect the payload, nor do they work on reading a set of messages from a different topic/partition than they were originally produced on.

See the help text for more information on how to use the commands.

### Development

A few notes as you read the code:

1. Any global state is attached to the context. This allows easy access to sharing state in a structured way that would allow for testing and prevents a proliferation of shared global variables between packages.
2. Everything is structured for graceful shutdown. Again, code should check the context to see if things are done when working in a loop or waiting. This allows for graceful shutdown. As we want ducktape to be in control of when the process exits so it can capture the latest state before shutting down. So there is an endpoint to shutdown the process. All this is wired through the context.
3. At the time of writing, this only supports verifying "mirror" or "noop" transforms, but over time we can perform more validation for other interesting kinds of transforms. When adding new validation, keep in mind that transforms ensure at-least-once delivery, so care should be taken to allow for these edge cases when duplicates are emitted, as they are valid.

---

### Overview (`tests/java/iceberg-rest-catalog/README.md`)

# Iceberg rest image

For converting different catalog implementations into a rest one.

## Run

To run it locally:

```sh
export CATALOG_CATALOG__IMPL=org.apache.iceberg.aws.glue.GlueCatalog

java -jar ./build/libs/iceberg-rest-image-all.jar
```

## Browse

To browse the catalog, you can use `pyiceberg`:

```
/* Detailed source-code truncated for AI context efficiency. */
```

## Building the Docker Image locally

If you want to make changes to the local files, and test them out, you can build the image locally and use that instead:

```bash
docker image rm -f tabulario/iceberg-rest && docker build -t tabulario/iceberg-rest .
```

---

### Overview (`tests/java/kafka-serde/README.md`)

# Java Kafka Serde Client

This repository contains a client that can be used to validate
the operation of Redpanda's Schema Registry via the Kafka Java library
provided by Confluent.

The application will create either a simple AVRO or PROTOBUF schema,
serialize a number of messages, produce them, and then consume and validate
that the message can be deserialized.

This test will exercise the schema registry by attempting to register and then
read the generated schema.

## Building

### Pre-requisites

* Java compiler
* Maven

### Build Instructions

```shell
mvn clean package --batch-mode --file <src>/tests/java/kafka-serde/ --define buildDir=<builddir>
```

#### Usage

```shell
usage: Java Kafka Serde Client [-h] -b BROKERS [-t TOPIC] -s SCHEMA_REGISTRY [-g CONSUMER_GROUP] [-p {AVRO,PROTOBUF}] [-c COUNT] [--security SECURITY]

Java Kafka Serde Client

named arguments:
  -h, --help             show this help message and exit
  -b BROKERS, --brokers BROKERS
                         Comma separated list of brokers
  -t TOPIC, --topic TOPIC
                         Name of topic (default: 16cd3248-b384-4042-bf1f-1b8b818a2fb5)
  -s SCHEMA_REGISTRY, --schema-registry SCHEMA_REGISTRY
                         URL of Schema Registry
  -g CONSUMER_GROUP, --consumer-group CONSUMER_GROUP
                         Consumer group (default: 605719f4-752f-4fc0-b3d5-8e7cc850ee41)
  -p {AVRO,PROTOBUF}, --protocol {AVRO,PROTOBUF}
                         Protocol to use (default: AVRO)
  -c COUNT, --count COUNT
                         Number of messages to send and consume (default: 1)
  --security SECURITY    JSON formatted security string
```

---

### Overview (`tests/models/cloud_topics/README.md`)

This is a P model for L0-only commit protocol, plus fetch. The architecture
covered by the model is depicted in the diagram below.

```
                                L0 Object                                   
                                    |                                        
                                    |                                        
                                    |                                        
                                    |                                        
                                    |                                        
                            +-------|------+                                 
                            |              |                                 
+------------+  Produce/Ack |              |                                 
|            | --------------              |   Placeholder append            
|   Client   |              |    Broker    ------------------------ Partition
|            | --------------              |                                 
+------------+     Fetch    |              |                                 
                            |              |                                 
                            +--------------+                                 
```

Notably absent from this model are:

* Reconciliation
* Leveling
* Recovery
* Garbage collection
* Failures

---

### Overview (`tests/rptest/chaos_tests/README.md`)

# Chaos tests

Ducktape port of the chaos test suite from https://github.com/redpanda-data/chaos/

These tests have a common scenario:
1. A redpanda cluster is started
2. A workload is started. Workload can be executed on one or several workers
   that write down their operations in a workload log.
3. After some warmup period a fault is introduced. Faults can be recoverable
   (they will be healed after some time) and oneoff.
4. Optionally we check if the workload made progress during the fault.
5. After some time the workload is stopped and its logs are copied to the control
   node and analyzed for consistency. Also some stats are calculated and
   throughput/latency graphs are plotted.

Original test set had the following structure:
* Each test case is a json file that specifies the scenario, the workload,
  the fault, and some additional checks and settings.
* Test cases are grouped into 8 test suites. All tests in a suite have the
  same scenario and workload but different faults.
* There are 5 scenarios and 7 workloads, i.e. some scenarios and workloads are
  reused in different workloads, but mostly it is a 1-1-1 scenario-workload-suite
  correspondence.

This is translated into the following ducktape test structure:
* Workloads are ducktape services inheriting from WorkloadServiceBase.
* Scenarios correspond to test classes. Each class inherits from the base
  scenario class, but they differ in how they prepare the cluster and topics.
* Suites correspond to test methods of the scenario classes.
* Each test case corresponds to a case_id parameter passed to a test method.
* The test method instantiates the workload and the fault (based on the
  case_id parameter) and runs the scenario.

---

### Overview (`tests/rptest/misc_tests/README.md`)

In this directory we have miscellaneous tests which don't run anywhere by
default, so they can include tests which fail, timeout, etc in order to
test the ducktape handling of these things (as they can be targeted directly
by a slash command or locally).

---

### Overview (`tests/rptest/remote_scripts/README.md`)

# remote_scripts

These scripts are copied to remote test nodes and
run there, via the inject_remote_script() function.

It's a convenient way of running some python on a test
node without having to build that python into our
test node image in advance.

---

### Overview (`tests/rptest/scale_tests/README.md`)

# scale_tests

Tests in this directory validate redpanda at higher
scale than the regular functional tests in tests/.

They require "fast" test nodes, where fast means dedicated
CPU/memory/storage for each node (not that the nodes necessarily
have to be especially high performance).

In practice, these are tests that we run using "clustered ducktape", i.e.
ducktape on EC2 instances, rather than running ducktape within docker.

---

### Overview (`src/transform-sdk/go/transform/README.md`)

# Golang Data Transforms SDK

This module contains a library for writing Data Transforms within Redpanda. It supports both [tinygo] and GOOS=wasip1 for standard [golang].

[tinygo]: https://tinygo.org/docs/guides/webassembly/wasi/
[golang]: https://go.dev/blog/wasi

---

### Overview (`src/transform-sdk/rust/core-sys/README.md`)

# Redpanda Data Transforms Rust SDK - ABI Contract

This is an internal crate as part of Redpanda's Data Transforms SDK, and encapsulates the broker's ABI contract with the guest.

If you are looking to use transforms you probably want crate [redpanda-transform-sdk](https://crates.io/crates/redpanda-transform-sdk).

---

### Overview (`src/transform-sdk/rust/core-types/README.md`)

# Redpanda Data Transforms Rust SDK - Shared Types

This is an internal crate as part of Redpanda's Data Transforms SDK, to provide shared types to other crates.

If you are looking to use transforms you probably want crate [redpanda-transform-sdk](https://crates.io/crates/redpanda-transform-sdk).

---

### Overview (`src/transform-sdk/rust/sr-sys/README.md`)

# Redpanda Data Transforms Schema Registry Client Rust SDK - ABI Contract

This is an internal crate as part of Redpanda's Data Transforms Schema Registry SDK, and encapsulates the broker's ABI contract with the guest.

If you are looking to use transform's Schema Registry client you probably want crate [redpanda-transform-sdk-sr](https://crates.io/crates/redpanda-transform-sdk).

---

### Overview (`src/transform-sdk/rust/sr-types/README.md`)

# Redpanda Data Transforms Schema Registry Client Rust SDK - Shared Types

This is an internal crate as part of Redpanda's Data Transforms Schema Registry SDK, to provide shared types to other crates.

If you are looking to use transform's Schema Registry client you probably want crate [redpanda-transform-sdk-sr](https://crates.io/crates/redpanda-transform-sdk).

---

### Overview (`src/transform-sdk/rust/sr/README.md`)

# Redpanda Data Transforms Schema Registry Client Rust SDK

This crate provides access to Redpanda's Data Transform's Schema Registry Client.

---

### Overview (`src/v/cloud_topics/frontend/README.md`)

This module contains components used in the kafka/data module.
This includes the log reader and the frontend components.
The reader implements 'model::log_reader::impl'. The 'frontend'
is used to implement a 'partition_proxy'.

---

### Overview (`src/v/pandaproxy/schema_registry/README.md`)

# Schema Registry

## References

- REST API: https://docs.confluent.io/platform/current/schema-registry/develop/api.html
- Alternative implementation: https://github.com/aiven/karapace

## High level design

Schema registry is built in to `pandaproxy`.

A single-partition topic `_schemas` is used as a backend store.

Each node maintains an in-memory copy of all schema data, lazily updated from the topic.

Active/Active HA: any node can serve reads and writes.

User-facing REST API is compatible with other products' schema registry implementations, but internal storage format is
not intended to be compatible.

*Future: we should apply a sanity limit on the number of schemas, as we hold them all in memory.*

## Writer synchronisation

**tldr; Optimistic writes, first writer wins**

Each write includes a sequence number in the key. This sequence number increases monotonically, and any write with an
out-of-order sequence number is ignored by readers. Writers set the sequence number for their next write as the most
recent message sequence they've seen, plus one.

While the sequence number is logically independent of the offset, it is very convenient to simply use the offset as the
sequence number:
that way, message validity can be checked by simply comparing the sequence number to the actual offset where the message
was written.

Under contention, writers know whether their write was successful by comparing the offset returned by their write
against the offset where they expected to write. If not equal, writer retries until they succeed.

Using the offset as the effective sequence number comes with a downside: concurrent writes from different nodes are more
expensive, as otherwise-valid writes can be rendered invalid by another /invalid/ write arriving before and knocking
them out of their intended offset. However, schemas are likely to be written at comparatively low rates, and usually by
a single API client (e.g. a CI system deploying new schemas for a new software build). This can be improved in future by
switching to a true sequence number (such that for a given sequence number under contended writes, at least one writer
is guaranteed to win).

Allocation of schema IDs and version IDs is done locally by writers based on the contents of their `ShardedStore`. If
this turns out to be out of date, then their write will retry (allocation is re-done on retry). Changes to the
`ShardedStore`, including advancing the next ID to allocate, are only applied once the write persists successfully.

To avoid compaction incorrectly preserving the last contended write (contended meaning same seq number) rather than
first write to a particular subject-version, the writer's node ID is also included in the key.

For permanent deletion of subjects or versions (i.e. writing tombstones), writers must remember the sequence number and
node ID that originally wrote to the subject, in order to regenerate the original write's key. These permanent deletion
writes are not subject to any sequence-ordering logic, and if multiple nodes perma-delete the same thing concurrently,
they will both succeed.

Q: Why is the sequence number global rather than per-subject?

- A: Because different subjects refer to the same schema IDs: if we treated subjects as independent, they might both
  grab the same schema ID for different schema definitions.

See worked examples further down that include sequence numbers.

## I/O

### Reads

Reads are served symmetrically from any node.

High traffic read endpoints are usually served from memory, falling back to read from the topic only if the requested ID
is not found locally (i.e. the local ShardedStore is out of date).

These fast-read endpoints are:

- `GET /schemas/ids/{int: id}`
- `GET /subjects/(string: subject)/versions/(versionId: version)`
- `GET /subjects/(string: subject)/versions/(versionId: version)/schema`

Endpoints that generate lists require a read from the topic to ensure that the list they are returning is complete.
These are:
GET /schemas/ids/{int: id}/versions
(because while we might have the schema in memory, it may have new subject-version associations we haven't seen yet)

- `GET /subjects`
- `GET /subjects/(string: subject)/version`
- `POST /subjects/(string: subject)`
- `GET /subjects/(string: subject)/versions/{versionId: version}/referencedby`

Within a single node, only one read to the topic is in flight at at time (enforced by a semaphore) -- this serialization
of reads reflects the serial layout of the underlying store.

*Future: use a queue-of-waiters synchronisation mechanism to reduce number of topic reads when many readers hit slow
endpoints concurrently within the same node*

*Future: if these endpoints turn out to be higher traffic than expected, introduce a config setting to make them
eventually consistent*

### Writes & deletes

As well as being subject to inter-node serialization via sequence numbers, writes are serialized within a node using a
semaphore. This simplifies the code, to avoid having rollback logic for projecting schema+version handle cases where
multiple writes are in flight & only some succeed.

*Future: if we find it necessary to

Writes are served symmetrically from any node, **but** it is advisable to send writes to the same node (doesn't matter
which)
as much as possible, to reduce the cost of synchronisation.

- `POST /subjects/(string: subject)/versions`

Soft deletions (without `permanent` parameter) are normal writes, which create new sequenced keys and include a flag in
the value to indicate that the named subject or version should be marked deleted.

- `DELETE /subjects/(string: subject)`
- `DELETE /subjects/(string: subject)/versions/(versionId: version)`

Permanent deletions render schemas unreadable, and are expected to free underlying storage. For us, that means emitting
tombstones to the keys used when writing the subject/version.

- `DELETE /subjects/(string: subject)?permanent=true`
- `DELETE /subjects/(string: subject)/versions/(versionId: version)?permanent=true`

### Examples

*Using shorthand key format of sequence-node-subject-version, where nodes are like A, B, C...*

#### Uncontended create & delete lifecycle

    # POST /subjects/foo
    1-A-foo-1 {schema}
    # POST /subjects/foo
    2-A-foo-2 {schema}
    # DELETE /subjects/foo
    3-A-foo-1 {deleted=true}
    4-A-foo-2 {deleted=true}
    # DELETE /subjects/foo?permanent=true
    1-A-foo-1 NULL
    1-A-foo-2 NULL
    3-A-foo-1 NULL
    4-A-foo-1 NULL

#### Contended writes (same subject+value)

    # POST A/subjects/foo
    1-A-foo-1 {schema}
    # Node A discovers they succeeded, returns schema=1 to caller

    # POST B/subjects/foo
    1-B-foo-1 {schema}
    # Node B re-reads, sees A's write, realizes value was already applied, returns schema=1 to caller

#### Contended writes (different subject)

    # POST A/subjects/foo
    1-A-foo-1 {schema}
    # Node A succeeds at offset 1.

    # POST B/subjects/bar
    1-B-foo-1 {schema}
    # Node B sees that its write landed at 'wrong' offset.  Re-reads.  Retries write.
    2-B-foo-1 {schema}
    # Node B succeeds at offset 2.

#### Contended hard deletes

    # DELETE A/subjects/foo?permanent=true
    1-A-foo-1 NULL

    # DELETE A/subjects/foo?permanent=true
    1-A-foo-1 NULL

    # Both nodes 'win'.

---

### Overview (`src/v/serde/json/README.md`)

# Seastar friendly JSON parser

`serde/json` provides a Seastar-friendly streaming JSON parser. Low level
parsing routines (string, numeric) are ported from [RapidJSON] and redesigned
for incremental parsing. As a result, the parser can be used to parse JSON
documents incrementally, i.e. the input buffer can be split into multiple chunks
with arbitrary alignment/boundaries.

StAX (Streaming API for XML) style API is used to parse JSON documents.

```cpp
auto p = serde::json::parser(iobuf);

while (co_await p.next()) {
    switch (p.token()) {
    case serde::json::token::object_start:
        // handle object start
        break;
    // ... handle other token types
    }
}
```

## Known issues

- Numeric precision lower than RapidJSON. While this can be ignored as
  [RFC 8259] does not mandate precision for numbers (section 6), we are still
  below what C++ double type can represent.
- Not sufficiently tested. Notably, no fuzz testing is done yet. The parser
  should be fuzz tested with various JSON documents to ensure it does not
  trip/gets into infinite loops with malformed documents. The parser should also
  be fuzz tested with various alignment/boundaries.
- No UTF-8 validation. The parser does not validate UTF-8 sequences.

## Design

### Push-parsers (e.g. string, numeric sub-parsers)

Push-parsers are our solution for asynchronous incremental parsing.

**Pros:**

- Can be fed with arbitrarily aligned buffers.
- Do not need to perform preemption checks if the buffers are sized properly
  (i.e., not too large).
- Do not require `co_await` for every byte read (e.g., if parsing is implemented
  from `seastar::input_stream`).
- For unicode strings the string parser can do zero-copy parsing (i.e., no need
  to copy the string into a separate buffer but will share the
  `seastar::temporary_buffer`).

**Cons:**

- Increased complexity.

While parsing from a `seastar::input_stream` is not supported yet, it is nice to
have primitives which allow us to implement it in the future without rewriting
the parsers.

## References

- [RFC 8259] - The JavaScript Object Notation (JSON) Data Interchange Format
- [RapidJSON] - A fast JSON parser/generator for C++ with both SAX and DOM style
  API

[rapidjson]: https://github.com/Tencent/rapidjson
[rfc 8259]: https://datatracker.ietf.org/doc/html/rfc8259

---

### Overview (`src/v/serde/parquet/README.md`)

# Parquet Library

Here lies a library to be able to write parquet files for Redpanda's Iceberg integration.
Due to Redpanda's usage of Seastar off the shelf preexisting parquet libraries do not meet
our strict requirements, imposed by our userland task scheduler and virtual memory avoiding 
allocator.

### Record Shredding

Parquet shreds records into a columnar format using the [same algorithm][twitter-parquet] as published in the [Dremel
paper][dremel]. A very helpful step by step explainer of the algorithm can be found [here][random-useful-wiki].
Our implmentation of this can be found in `shredder.cc`.

[random-useful-wiki]: https://github.com/julienledem/redelm/wiki/The-striping-and-assembly-algorithms-from-the-Dremel-paper#step-by-step
[twitter-parquet]: https://blog.x.com/engineering/en_us/a/2013/dremel-made-simple-with-parquet
[dremel]: https://static.googleusercontent.com/media/research.google.com/en//pubs/archive/36632.pdf

### Metadata

Parquet metadata is serialized using [Apache Thrift's compact wire format][thrift-compact-format].

We use metadata that is the logical representation of what our application needs, then we write out
the wire format with all the deprecated and legacy types to be compatible with legacy query systems.

The physical format of serialized parquet metadata is documented [here][parquet-thrift].


[parquet-thrift]: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift
[thrift-compact-format]: https://github.com/apache/thrift/blob/master/doc/specs/thrift-compact-protocol.md

---

### Overview (`src/v/serde/protobuf/README.md`)

# Protobuf Parser

This directory contains a seastar friendly protobuf parser.
It should adhere to the same compatibility guarantees as the offical C++ library
but has a few notable differences:

1. Does not make contiguous allocations of repeated fields, maps or strings/bytes types
2. Is reactor friendly on deeply nested or large protobufs in that it will yield control
3. Is a stackless parser, so it is not bound by the smallish 1MB stacks that seastar uses for threads

## Development

If you are tasked with updating this code, here are a few helpful links:

1. [Encoding spec](https://protobuf.dev/programming-guides/encoding/) (note this elides some important details about how invalid/corrupted data is handled)
2. [Golang protobuf parser](https://github.com/protocolbuffers/protobuf-go/blob/master/proto/decode.go)
3. [Java protobuf parser](https://github.com/protocolbuffers/protobuf/tree/main/java/core/src/main/java/com/google/protobuf)
4. [C++ protobuf parser](https://github.com/protocolbuffers/protobuf/blob/main/src/google/protobuf/wire_format_lite.cc)
5. [Protobuf Zero](https://github.com/mapbox/protozero/blob/master/include/protozero/pbf_reader.hpp)

---

### Overview (`src/v/serde/thrift/README.md`)

# Apache Thrift Library

Here is an implementation of [Apache Thrift's compact wire format][thrift-compact-format]. Only what is needed for our parquet support is implementated here, which at the time of writing is encoding only support for a structs, lists, strings and numbers.

[thrift-compact-format]: https://github.com/apache/thrift/blob/master/doc/specs/thrift-compact-protocol.md

---

### Overview (`src/v/test_utils/go/README.md`)

# Golang Verification Programs

Sometimes it's useful for tests to verify behavior against another library implementation (kafka protocol, serialization format, etc).
These golang programs are used in verification of Redpanda, and all share a single `go.mod` to limit the sprawl of `go.mod` files in
Redpanda.

---

### Overview (`src/v/transform/logging/README.md`)

# Transform Logging Library

This module is responsible for collecting logs coming from writes on stdout/stderr from the Wasm virtual machine. These logs sit in a (configurable) buffer size per transform, then we batch them up and write them to the `_redpanda.transform_logs`. Writes happen by using the transform data path RPCs to produce logs. The major entry point to this subsystem is the `log_manager`, which does the routing of logs and collecting into buffers.

---

### Overview (`src/v/transform/rpc/README.md`)

# Transform RPC Libary

Data Transforms has it's own set of RPCs for both the data path of writing transform output, as well as various control plane actions. To use these RPCs, there is a client available that gives a nicer interface to these RPCs.

## Data Plane RPCs

These are the RPCs used on the data plane, to understand more see the documentation in `client.h`.

* produce
* offset_fetch
* batch_offset_commit
* find_coordinator

## Control Plane RPCs

These are the RPCs used on the control plane, to understand more see the documentation in `client.h`.

* store_wasm_binary
* delete_wasm_binary
* load_wasm_binary
* generate_report
* list_committed_offsets
* delete_committed_offsets

## Client implementation

The client takes each operation to be performed and sees if this node can service the request or if
it needs to make the request on the network. If it can service the request on this node, then the
request is directly forwarded to the local service without a network hop, otherwise it uses our RPC
framework to make the request on the appropriate node. Both kinds of requests are automatically retried
for specific error codes.

## Service implementation

The service handles interacting with the local system as a whole on behalf of the transform subsystem.
For fetch/produce data plane actions these look similar to our kafka handlers, while other actions will
directly interact with the correct subsystem to perform actions (such as getting the offset STM).

## System dependencies

All of the client/service is able to be unit tested, which is handled by creating interfaces for the
rest of the system. In production usage we have implemented these interface uses the appropriate cluster
component (controller, partition manager, etc), while in unit tests we inject fake implementations.

---

### Overview (`src/v/transform/stm/README.md`)

# Transform State Machine Library

This very small library holds the template instantiation of the distributed key value store for data transform's offset commit data.

---

### Overview (`src/v/wasm/parser/README.md`)

# WebAssembly Parser Library

The Wasm parser library contains routines to parse WebAssembly in the binary format. This can be used to do static analysis of WebAssembly modules as they are deployed.
We only support what is defined in the WebAssembly v2 Core Spec. We do not support any proposals that have not been merged into the spec. We may choose to add some
support for extra functionality as we make use of it in the Wasm subsystem. Lastly, this is not a full parser, but only parses the "API" so to speak of a module. Specifically
that is imports and exports of the defined types and excludes parsing instructions (function bodies).

If you're looking to develop this library or make changes the following links maybe helpful to understand WebAssembly's binary format:

* [Offical WebAssembly Core Specification](https://webassembly.github.io/spec/core/)
* [Offical WebAssembly Testsuite](https://github.com/WebAssembly/testsuite)
* [WebAssembly Proposals](https://github.com/WebAssembly/proposals)
* [Understanding Every Byte in a WASM Module](https://danielmangum.com/posts/every-byte-wasm-module/)

---

### Overview (`tests/rptest/clients/admin/README.md`)

## Redpanda Admin API Client

Python ConnectRPC client for the admin v2 API, used by ducktape tests. The proto bindings are generated and committed; regenerate them after any change under `proto/redpanda/core/admin/**`:

```
./tools/regenerate_ducktape_protos.sh
```

For the end-to-end process of adding a new admin v2 endpoint (proto authoring, C++ service impl, Bazel wiring, and the regen step above), use the `add-admin-v2-endpoint` skill.

---

### Overview (`proto/redpanda/core/admin/internal/README.md`)

# Internal Redpanda RPC endpoints

The endpoints that live in this directory have no backwards compatibility guarentees (or really any external guarentees), outside of the Core Redpanda engineering team.

---

### Overview (`src/go/rpk/pkg/tuners/ReadMe.md`)

# Redpanda Keeper Performance Tuners

This package contains all performance tuners that are used by rpk to adjust Linux to achieve the best performance while using Redpanda.

## Supported Tuners

The following tuners are supported

### Disk IRQs

The disk IRQs tuner binds all disk IRQs to requested set of CPUs. This tuner uses `hwloc` library to compute CPU masks. Prevent IRQ Balance from moving tuned devices IRQs. CPU set that is used by the tuner can be limited by CPU mask parameter. If mask parameter is provided then only those CPUs that are masked will be considered as available. Mask covering all host CPUs is used as a default.

IRQs are distributed according to the following rules:

- Distribute NVMe disks IRQs equally among all available CPUs.
- Distribute non-NVMe disks IRQs equally among designated CPUs or among all available CPUs in the `mq` mode.

#### Modes description

 `sq` - set all IRQs of a given disk to CPU0

 `sq_split` - divide all IRQs of a given disk between CPU0 and its HT siblings

 `mq` - distribute disk IRQs among all available CPUs instead of binding them all to CPU0

  If there isn't any mode given script will use a default mode:
     - If there are no non-NVMe disks the `mq` mode will be used
     - For non-NVMe default mode will be set to `mq` if number HT siblings is less or equal to 4 otherwise if number of cores is less or equal to 4 the `sq` mode will be used in any other conditions the `sq-split` will be the default

### Disks Scheduler

The disk scheduler tuner set the block device I/O scheduler in order to minimize the overhead of kernel I/O scheduler. Preferred scheduler is of `none` type - it istruct kernel not to use I/O scheduler and use `blk-mq` module instead (there will be one queue per CPU), if the one is not supported `noop` will be used.

### Network Tuner

The Network IRQs tuner binds all NIC IRQs to requested set of CPUs. This tuner uses `hwloc` library to compute CPU masks. Prevent IRQ Balance from moving tuned devices IRQs. CPU set that is used by the tuner can be limited by CPU mask parameter. If mask parameter is provided then only those CPUs that are masked will be considered as available. Mask covering all host CPUs is used as a default.

As a result some of the CPUs may be destined to only handle the IRQs and taken out of the CPU set that should be used to run the seastar application ("compute CPU set"). 

#### Modes description

 `sq` - set all IRQs of a given NIC to CPU0 and configure RPS
      to spreads NAPIs' handling between other CPUs - should be used as default for low end NICs

 `sq_split` - divide all IRQs of a given NIC between CPU0 and its HT siblings and configure RPS
      to spreads NAPIs' handling between other CPUs

 `mq` - distribute NIC's IRQs among all CPUs instead of binding
      them all to CPU0. In this mode RPS is always enabled to
      spreads NAPIs' handling between all CPUs - should be used as default for server workloads and high throughput NICs

 If there isn't any mode given script will use a default mode:
    - If number of physical CPU cores per Rx HW queue is greater than 4 - use the 'sq-split' mode.
    - Otherwise, if number of hyperthreads per Rx HW queue is greater than 4 - use the 'sq' mode.
    - Otherwise use the 'mq' mode.

---

### Overview (`src/transform-sdk/rust/examples/jaq/README.md`)

# jaq transform

[`jaq`][jaq] is a Rust clone of the popular JSON processing tool `jq`.

This example transform supports defining an environment variable `FILTER` that defines a `jaq` filter, which is applied to all record values.

## Example Usage

```
cargo build --release --target=wasm32-wasip1
rpk transform deploy ../../target/wasm32-wasip1/release/jaq.wasm \
    --name=email-redaction-transform \
    --input-topic=$MY_INPUT_TOPIC \
    --output-topic=$MY_OUTPUT_TOPIC \
    --var 'FILTER=del(.email)'
echo '{"data":9,"email":"foo@bar.com"}' | rpk topic produce $MY_INPUT_TOPIC
rpk topic consume $MY_OUTPUT_TOPIC
# {"data":9}
```

[jaq]: https://github.com/01mf02/jaq

---

### Overview (`src/v/cloud_topics/level_zero/reader/README.md`)

### L0 read path

The read request is first added to the `read_pipeline` component. The `resolver` component fetches
read requests and resolves them by creating the reader that returns materialized record batches.

The record batches are either fetched from the cloud storage or cache (either in-memory or disk).
Current implementation is not doing any caching or multi-step processing. Instead, it just creates a
reader object and resolves the read request using this reader.

The resolver is gated by the requested offset range. If the offset range is above the last reconciled
offset or it contains last reconciled offset the resolver is invoked. Otherwise L1 resolver should be
invoked.

---

### Overview (`src/v/iceberg/tests/testdata/README.md`)

# Iceberg test data

Data generated for use in tests.

```
RP_HOME=$(git rev-parse --show-toplevel)
uv run $RP_HOME/src/v/iceberg/tests/gen_test_iceberg_manifest.py -o $RP_HOME/src/v/iceberg/tests/testdata/nested_manifest.avro -n 100
```

---

### Overview (`src/v/test_utils/go/kreq-gen/README.md`)

# kafka-request-generator

The kafka-request-generator is a small little go program that generates 
random kafka requests (or responses) and writes their serialized representation to stdout. It depends on 
[franz-go](https://github.com/twmb/franz-go) to do this.


``` sh
➜  workspace/scratch ./kafka-request-generator -h
Usage of ./kafka-request-generator:
  -api int
        API key of request to generate
  -binary-output
        Print result as binary (default true)
  -is-request
        Generate kafka request if true, response if false (default true)
  -version int
        API version of generated request

```

Pass the kafka api request key and the api version of a request you want to 
generate. If its actually a response pass `-is-request=false` to the program. 
It will write binary data to stdout, so if your using this for manual testing 
and would like to see ASCII data on stdout instead for whatever reason, pass 
-binary-output=false to the program to get a hex formatted represetation 
instead.

### Generated requests

The requests generated are requests with fields populated with random data 
based on the type of a given field. For example if a field representing an 
enum is defined as type `int8` in the request schema but there are only 3 
valid values for said enum, all valid values for an `int8` could be possible 
results the generator would populate the field with.

The only assumptions that can be made about the contents of a given field are 
the binary data fields that are part of produce and fetch requests that 
represent record batches. Properly formatted serialized 
`kafka::record_batch`es (or legacy batches for older versions) are written to 
those fields.

### Example

``` sh
➜  workspace/scratch ./kafka-request-generator -api 16 -version 9 -is-request=false -binary-output=false
241a39ed1147021651644c4b73774a50466f5458755648794b4364505a0d6f685269696b4a4b7
75a4c4e206669565267684c6b75747a456e51734c456a4e4f787a7a5852586b4d4f704a000164
10d4ddb038cef005c2fb1c8a1731966e45%
```

## Current usage

Currently this mainprog is invoked within a particular C++ unit test (located 
in src/v/kafka/protocol/tests/protocol_test.cc) for ensuring that redpanda is 
binary compatible with all currently supported api/versions of the kafka 
protocol. Note that this only means that redpanda can correctly serialize / 
deserialize a request to spec, it does not imply that a particular client at 
a version can or cannot expect some sort of behavior from the broker.

---

### Overview (`src/transform-sdk/go/transform/internal/testdata/README.md`)

This directory contains a collection of transforms using for Redpanda internal testing.

---

### Overview (`src/v/schema/protobuf/google/type/README.md`)

# Google Common Types

This package contains definitions of common types for Google APIs.
All types defined in this package are suitable for different APIs to
exchange data, and will never break binary compatibility. They should
have design quality comparable to major programming languages like
Java and C#.

NOTE: Some common types are defined in the package `google.protobuf`
as they are directly supported by Protocol Buffers compiler and
runtime. Those types are called Well-Known Types.

## Java Utilities

A set of Java utilities for the Common Types are provided in the
`//java/com/google/type/util/` package.

---

### Overview (`src/go/rpk/pkg/cli/generate/graf/README.md`)

# graf

`graf` is a Go representation of the
[Grafana Dashboard JSON model][grafana-json-model].

Its main reason is to allow building and generating custom dashboards in Go and
exporting them to JSON so that they can be imported in Grafana.

Currently supported panels: row, graph, text, singlestat.

There are constructors for all the panels, which provide sensible default values
for each type & sequential ID assignment.

## Usage

```
/* Detailed source-code truncated for AI context efficiency. */
```

## Things to keep in mind

### Grid system

As seen in the example above, it's up to you to set each panel's
[GridPos][grafana-grid] `X` & `Y` values. If you don't, Grafana will put all the
panels in the same coordinates ((0, 0) by default) and the dashboard won't work.
You can read more about Grafana's grid system in the
[Grafana docs][grafana-grid].

### Panel IDs

`graf` assigns an auto-incremented ID to each panel if it's created through a
constructor (e.g. `NewGraphPanel`). However, you're free to use your own IDs,
or to create the panels directly. In that case, make sure you assign a unique ID
to each; Grafana doesn't auto-assign panel IDs when a dashboard is imported, and
it doesn't support 2 panels having the same ID, which causes the whole dashboard
to break.

It's also worth noting that IDs with very high values may break the dashboard
when you import it, so limit their range if you're assigning random IDs.


## Similar libs
- [https://github.com/grafana-tools/sdk](grafana-tools/sdk)


[grafana-grid](https://grafana.com/docs/grafana/latest/reference/dashboard/#panel-size-and-position)

---

### Overview (`src/v/serde/json/tests/testdata/json-test-suite/README.md`)

Test suite from https://code.google.com/archive/p/json-test-suite/downloads

Distributed under Apache License 2.0 by Yidong Fang.

---

### Overview (`src/v/serde/json/tests/testdata/jsonchecker/README.md`)

Test suite from http://json.org/JSON_checker/.

If the JSON_checker is working correctly, it must accept all of the pass*.json files and reject all of the fail*.json files.

Changes:
- The json_checker_test uses a depth limit of 19 to ensure fail18.json (20 levels) fails while pass2.json (19 levels) succeeds.
- fail1.json: REMOVED - Originally rejected top-level strings, but updated JSON spec allows any value at top level.

---

### Overview (`src/go/rpk/pkg/cli/transform/template/golang/README.md`)

# Redpanda Golang WASM Transform

To get started you first need to have at least go 1.20 installed.

You can get started by modifying the <code>transform.go</code> file
with your logic.

Once you're ready to test out your transform live you need to:

1. Make sure you have a container running via <code>rpk container start</code>
1. Run <code>rpk transform build</code>
1. Create your topics via <code>rpk topic create</code>
1. Run <code>rpk transform deploy</code>
1. Then use <code>rpk topic produce</code> and <code>rpk topic consume</code>
   to see your transformation live!

---

### Overview (`src/go/rpk/pkg/cli/transform/template/js/README.md`)

# Redpanda %s WASM Transform

To get started you first need to have node and npm installed.

You can get started by modifying the <code>src/index.%s</code> file
with your logic.

Once you're ready to test out your transform live you need to:

1. Make sure you have a container running via <code>rpk container start</code>
1. Run <code>rpk transform build</code>
1. Create your topics via <code>rpk topic create</code>
1. Run <code>rpk transform deploy</code>
1. Then use <code>rpk topic produce</code> and <code>rpk topic consume</code>
   to see your transformation live!

---

### Overview (`src/go/rpk/pkg/cli/transform/template/rust/README.md`)

# Redpanda Rust WASM Transform

To get started you first need to have at least stable rust 1.72 installed.

You can get started by modifying the <code>src/bin.rs</code> file
with your logic.

Once you're ready to test out your transform live you need to:

1. Make sure you have a container running via <code>rpk container start</code>
1. Run <code>rpk transform build</code>
1. Create your topics via <code>rpk topic create</code>
1. Run <code>rpk transform deploy</code>
1. Then use <code>rpk topic produce</code> and <code>rpk topic consume</code>
   to see your transformation live!

---

### Overview (`src/go/rpk/pkg/serde/embed/protobuf/google/type/README.md`)

# Google Common Types

This package contains definitions of common types for Google APIs.
All types defined in this package are suitable for different APIs to
exchange data, and will never break binary compatibility. They should
have design quality comparable to major programming languages like
Java and C#.

NOTE: Some common types are defined in the package `google.protobuf`
as they are directly supported by Protocol Buffers compiler and
runtime. Those types are called Well-Known Types.

## Java Utilities

A set of Java utilities for the Common Types are provided in the
`//java/com/google/type/util/` package.

---

### Writing High Performance Coroutine Code (`docs/writing-high-performance-coroutine-code.md`)

# TLDR:

 - Write sync code if possible
 - Use coroutines over continuations
 - Don't overoptimize for allocs in microbenchmarks unless the per coroutine
   work is very low, e.g.: when parsing fields on a per record level

# Writing high performance coroutine code

## Keep coroutine chains intact aka. just use coroutines

HALO (heap allocation elision optimization) can merge nested coroutine frames
into the parent frame. This works best if all code is coroutines. coroutine ->
continuation -> coroutine -> continuation doesn't allow for frame merging. Note
that there is no magical optimizer that will help with this. 

**Do:** Keep async call chains coroutine based all the way down.

```cpp
ss::future<> a() {
  co_await b();
}

ss::future<> b() {
  co_await c();
}
```

**Don't:** Mix coroutines and continuations 

```cpp
ss::future<> a() {
    co_await b().then(...)
}
```

## Await futures directly

If the returned future of a coroutine is not needed for async purposes
`co_await` a coroutine directly. Otherwise HALO won't apply.

**Do:** 

```cpp
co_await foo();
```

**Don't:** 

```cpp
auto f = foo();
co_await std::move(f);
```

## Pass futures directly

`when_all`, `when_all_succeed`, `coroutine::as_future`, `coroutine::try_future`
and `coroutine::without_preemption_check` are HALO aware.

**Do:** Pass newly created coroutine futures directly to these wrappers.

```cpp
co_await ss::when_all(a(), b());
co_await ss::coroutine::as_future(c());
```

**Don't:** Store the child future first when it can be passed directly.

```cpp
auto f = c();
co_await ss::coroutine::as_future(std::move(f));
```

## Be aware of scheduling semantics

The biggest functional and also performance difference between continuations and
coroutines is their scheduling semantics.

`co_await` is a preemption point.

`.then` doesn't perform the same preemption check and future returning callbacks
use Seastar's urgent scheduling (in contrast to value returning callbacks that
use normal append-to-end-of-taskqueue scheduling).

**Do:** Make an intentional choice when replacing continuation code.

```cpp
co_await ss::coroutine::without_preemption_check(work());
update();
```

**Don't:** Assume a direct rewrite is scheduling equivalent. This can change
lock hold time, batching and latency.

```cpp
future<> work();

future<> cont() {
    return get_lock().then([] {  return work(); });
}

// not-equivalent
future<> coro() {
    auto lock = co_await get_lock();
    co_await work();
}
```

The coroutine version can potentially suspend twice, once just after acquiring
the lock and a second time before calling `work`.

The continuation version never suspends under lock. This isn't just about
correctness (most of the time it's fine to suspend under the lock) but also a
performance one.

Note that neither is directly better or worse. Suspending under the lock is
likely bad for latency but can cause batching in other parts of the system. This
is in all likelihood good for throughput and reduced reactor util. If you want
batching it's best to implement an explicit batching pattern that doesn't
rely on scheduling semantics.

## Don't overoptimize for allocations

Don't obsess over allocations related to coroutines and continuations. They only
matter if the per coroutine/continuation work is very small. There are very few
cases in Redpanda that fit this pattern, examples are: field parsing in
iceberg/protobuf/json/avro, per field compression handling etc. In the big
picture it makes very little difference CPU wise as most of the time the
work-per-coroutine is high.

LLMs love reducing allocs via coroutine/continuation changes as the allocs
number in microbenches is stable and via such optimizations is easy to reduce
(at the cost of readability/clarity).


## Write HALO aware wrappers

Constructor arguments don't currently propagate a safe elide context.

**Do:** Use a factory function and consider `coro_await_elidable_argument` when
the child is guaranteed to finish before the wrapper does.

```cpp
auto handle(SEASTAR_CORO_AWAIT_ELIDABLE_ARGUMENT ss::future<>&& f) {
    ...
    co_await f;
    ...
}
```

**Don't:** Add the attribute if the child can escape the wrapper. It is a
lifetime contract, not just an optimization hint.

```cpp
auto detach(SEASTAR_CORO_AWAIT_ELIDABLE_ARGUMENT ss::future<>&& f) {
    ...
    background(std::move(f)); // Wrong: f escapes the wrapper.
    ...
}
```

Again keep in mind that this level of tuning is only relevant for very commonly
used helper functions.

## Avoid recursive coroutine chains

The compiler can't embed a potentially unbounded number of frames.

**Do:** Use an iterative coroutine where practical.

```cpp
ss::future<> visit(node* n) {
  for (; n != nullptr; n = n->next) {
    co_await visit_one(n);
  }
}
```

**Don't:** Expect HALO to optimize recursive coroutine calls.

```cpp
ss::future<> visit(node* n) {
  if (n == nullptr) {
    co_return;
  }

  co_await visit_one(n);
  co_await visit(n->next);
}
```


## Yield in CPU heavy loops

**Do:** Use `coroutine::maybe_yield` to remain reactor friendly.

```cpp
for (const auto& item : items) {
  process(item);
  co_await ss::coroutine::maybe_yield();
}
```

**Don't:** Run an unbounded loop without a scheduling point.

```cpp
for (const auto& item : unbounded_input) {
  process(item);
}
```

Note that this is often the preferred way if it allows to keep (leaf) functions coroutine free (sync code is still the fastest).

## References

- [Coroutines Optimization Opportunities from Compiler's perspective](https://chuanqixu9.github.io/c++/2026/03/27/C++20-Coroutines-from-compiler-and-library-authors-perspective.en.html)
- [Clang `coro_await_elidable`](https://clang.llvm.org/docs/AttributeReference.html#coro-await-elidable)
- [Clang `coro_await_elidable_argument`](https://clang.llvm.org/docs/AttributeReference.html#coro-await-elidable-argument)
- [Clang `coro_only_destroy_when_complete`](https://clang.llvm.org/docs/AttributeReference.html#coro-only-destroy-when-complete)

---

### Writing Scalable And High Performance Code (`docs/writing-scalable-and-high-performance-code.md`)

# Writing scalable and high performance code

## Scale

Redpanda runs at scale and it is good to keep this in mind when writing code.
Most tests do not run at high scale, and the scale tests that do run have
limited coverage. So we cannot expect all scale issues to be caught by testing
and proactive scale-sensitive choices can go a long way towards mitigating that.

Some specific things to keep in mind when writing for scale, such as what data
structure to use, are very important, and these are addressed in subsections
below.

### Memory allocation limits (128K rule)

**As a guideline, do not allocate more than 128 KiB of contiguous memory in
redpanda after startup.**

Allocations larger than 128 KiB have a relatively high likelihood of failing
allocation even when substantial free memory is available due to
seastar-specific fragmentation. These failures won’t usually occur in our tests
(except perhaps very long running ones), but will fail at some inopportune
moment at an important customer.

128 KiB is not a magic number that it is guaranteed to work, but it is small
enough that we *believe* it won’t be a significant problem even in fragmented
heaps and it aligns with the existing sizes of some very common allocations such
as `iobuf` fragments.

The most common causes of large contiguous allocations are large arrays or array
like containers (such as `std::vector`, or the use of containers that are backed
by those types of containers, such as `std::unordered_map` or
`absl::flat_hash_map`).

#### Problematic containers

Containers below make large contiguous allocations and should be avoided if the
container size could reasonably exceed the 128K rule.

| **Container**                                       | **Approximate contiguous allocation size** | **Notes**                                                                                                                                                                                                                                                                                                                      |
| --------------------------------------------------- | ------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `std::vector<T>`                                    | `2 * size() * sizeof(T)`                   | Factor of two comes from the resize step: the size doubles so will contain `2 * size()` at that point. In some cases you remove this factor of two when you know the final size and use `reserve()` or `resize()`.                                                                                                             |
| `std::unordered_map<K, V>`, `std::unordered_set<K>` | `16 * size()`                              | Factor of 16 comes from `2 * sizeof(void *)` where 2 is due to the doubling during growth (so half the buckets will be empty) and `sizeof(void *)` reflects that these are closed-addressed hash maps which have a single pointer per bucket. The size of the contiguous allocation does not depend on the size of `K` or `V`. |
| `absl::flat_hash_map<K, V>`                         | `2.3 * (sizeof(K) + sizeof(V)) * size()`   | The contiguous allocation comes from the key/value array, as the separate control array is generally much smaller. The factor of 2.3 comes from ~1.15 size → capacity ratio (due to load factor) times 2 for doubling.                                                                                                         |
| `absl::node_hash_map<K, V>`                         | `2.3 * (8 + 1) * size()` (? untested)      | Similar to `std::unordered_map`, except an additional 1 byte per slot (the `+ 1` part) for control byte, and a `1.15x` factor due to load factor.                                                                                                                                                                              |

If all else fails (e.g., a third party lib really wants big contiguous
allocations), make the allocations at startup, when large contiguous blocks are
easy to obtain.

### Preferred datastructures:

Prefer these data structures as they are resistant to oversized allocations and
memory fragmentation.

| **Datastructure** | **Preferred**      | **Caveats**                                                                                              | **Alternatives**                                                                                                                                                                                                                                                                                                                                                   |
| ----------------- | ------------------ | -------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| Array, vector     | `chunked_vector`   |                                                                                                          | `std::deque`: Fine from oversized allocation perspective. The chunk size cannot be controlled. As soon as 1 element is added, the full chunk is allocated, making it a memory hog in scenarios where you have one per X (for example partition, topic etc.) as the overhead will be large.`boost::deque`: as above, but at least the chunk size can be controlled. |
| Unordered Map     | `chunked_hash_map` | Insert and erase invalidates all iterators: Don’t do `map.erase(it++)`, instead do `it = map.erase(it)`. | `absl::node_hash_map`: Useful if pointer stability is needed (try working around first). Don't use if scaling with partitions etc as this will suffer from oversized allocs                                                                                                                                                                                        |
| Ordered Map       | `absl::btree_map`  |                                                                                                          | `std::map`: Use if pointer stability is needed (probably can work around that by using std::unique_ptr values in the map)                                                                                                                                                                                                                                          |

### Managing Concurrency in the system

Seastar makes it very easy to write async and concurrent code and thus hide IO
latency. This allows for writing high performing code.

At the same time care needs to be taken to not introduce too much concurrency
into the system. Each concurrent task comes with memory usage that depending on
the task can be relatively large and if many of those tasks are in the system
waiting it can lead to OOM.

Think about how much concurrency is needed to adequately hide latency. For
sending large chunks of data to a local service with low latency not much
concurrency is needed, e.g.: uploading chunks of 128KiB to local S3 you can get
very good throughput on a single connection so not much concurrency is needed.
Writing 4KiB blocks to EBS will need higher concurrency to get good throughput.

Keep in mind that effectively most things are shard local so often higher
concurrency is already induced by having multiple shards.

There is also various patterns to manage concurrency and memory usage.

#### Limit the concurrency for a certain operation

Use `ss::max_concurrent_for_each` to limit max concurrency. Concurrency larger
than 100 is probably never a good idea, most of the time 10 is probably
approximately right but there is no hard rule. Keep above advice in mind.

Avoid using `ss::parallel_for_each` for ranges that can grow large such as
segments, topics or partitions.

Be aware of implicitly nested `ss::max_concurrent_for_each` usages. If it’s a
concern use a `ssx::semaphore` to better limit active concurrency.

#### Limit memory usage for a certain class of operations

An alternative approach if tasks themselves are not an issue is to limit memory
usage directly. We generally do this by using a `ssx::semaphore` which counts
the total available memory for this operation. Tasks try to draw from this
before they are allowed to execute and otherwise have to wait.

This is generally the better way to limit concurrency but can be hard to limit
if it’s hard to estimate how much memory a task will need or the memory need
isn’t known a-priori at all. Be careful in those cases as task throughput can be
starved if memory estimation is wrong.

## Performance

This section is a stub.

Unlike scale, you don’t need to design most of the code you write for the
highest performance (i.e., doing the required job as fast as possible). Most
code is cold or very cold. Even in hot paths, simple, straightforward code
should be the default.

Only in specific cases where profiling information shows a hotspot, or you have
solid reasoning to believe you are writing hot code, then some deviation from
“simple, straightforward” might be appropriate.

That disclaimer aside, here is a short list of (unsorted) performance notes:

### Miscellaneous Performance Notes

#### iobuf is not super cheap to copy or share

`iobuf` can be copied in two different ways: `iobuf::copy()` which creates a
byte-wise copy of all underlying buffers, or `iobuf::share()` which creates a
“logical” copy that shares the underlying buffers and so avoids a byte-wise
copy. Both of these can be relatively expensive: `copy()` in the obvious way
that it effective needs to do a `memcpy` of the entire buffer, and `share()`
because it needs to do an atomic operation and an allocation (though this is
O(1) work, so less relevant for larger buffers).

This means that *if* you have a sync function which takes an `iobuf` and does
not mutate it, just take it as a `const iobuf&` instead of by-value (followed by
moves) which is a pattern which occurs often in our codebase.

---

### Rfcs 00000000 Template (`docs/rfcs/00000000_template.md`)

- Feature Name:
- Status: draft/in-progress/completed/rejected/obsolete/postponed
- Start Date: YYYY-MM-DD
- Authors:
- Issue: (one or more # from the issue tracker)

# Executive Summary

One paragraph - High level explanation of the proposed change.

## What is being proposed

One paragraph

## Why (short reason)

One paragraph

## How (short plan)

One paragraph

## Impact

Long form.

# Motivation
## Why are we doing this?

One paragraph

## What use cases does it support?

One paragraph

## What is the expected outcome?

One paragraph

# Guide-level explanation
## How do we teach this?

Long form.

Explain the proposal as if it was already included in the project and
you were teaching it to another  programmer. That generally means:

## Introducing new named concepts.

- Explaining the feature largely in terms of examples. Take into account that
  a product manager (PM) will want to connect back the work introduced by
  the RFC with user stories. Whenever practical, do ask PMs if they already
  have user stories that relate to the proposed work, and do approach PMs
  to attract user buy-in and mindshare if applicable.
- Explaining how  contributors and users should think about
  the feature, and how it should impact the way they use
  . It should explain the impact as concretely as possible.
- If applicable, provide sample error messages, deprecation warnings, or
  migration guidance.
- If applicable, describe the differences between teaching this to
  existing pandas and new pandas.

For implementation-oriented RFCs (e.g. for core internals), this
section should focus on how contributors should think about
the change, and give examples of its concrete impact. For policy RFCs,
this section should provide an example-driven introduction to the
policy, and explain its impact in concrete terms.

# Reference-level explanation

( This is the technical portion of the RFC )

## Interaction with other features

- It covers where this feature may be surfaced in other areas of the product
   - If the change influences a user-facing interface, make sure to preserve
     consistent user experience (UX). Prefer to avoid UX changes altogether
     unless the RFC also argues for a clear UX benefit to users. If UX has
     to change, then prefer changes that match the UX for related features,
     to give a clear impression to users of homogeneous CLI / GUI elements.
     Avoid UX surprises at all costs. If in doubt, ask for input from other
     engineers with past UX design experience and from your design department.

## Telemetry & Observability

- It considers how to monitor the *success* and quality of the feature.
   - Your RFC must consider and propose a set of metrics to be collected,
     if applicable, and suggest which metrics would be useful to users and
     which need to be exposed in a public interface.
   - Your RFC should outline how you propose to investigate when users run
     into related issues in production. If you propose new data structures,
     suggest how they should be checked for consistency. If you propose new
     asynchronous subsystems, suggest how a user can observe their state via
     tracing. In general, think about how your coworkers and users will gain
     access to the internals of the change after it has happened to either
     gain understanding during execution or troubleshoot problems.
- It is reasonably clear how the feature would be implemented.

## Corner cases dissected by example.

- List all corner cases:
  And a detailed explanation of the corner cases

## Detailed design - What needs to change to get there

The section should return to the examples given in the previous
section, and explain more fully how the detailed proposal makes those
examples work.


## Detailed design - How it works

Describe the overview of the design, and then explain each part of the
implementation in enough detail that reviewers will be able to
identify any missing pieces. Make sure to call out interactions with
other active RFCs.

## Drawbacks

Why should we *not* do this?

If applicable, list mitigating factors that may make each drawback acceptable.

Investigate the consequences of the proposed change onto other areas of.
If other features are impacted, especially UX, list this impact as a reason
not to do the change. If possible, also investigate and suggest mitigating
actions that would reduce the impact. You can for example consider additional
validation testing, additional documentation or doc changes,
new user research, etc.

Also investigate the consequences of the proposed change on performance. Pay
especially attention to the risk that introducing a possible performance
improvement in one area can slow down another area in an unexpected way.
Examine all the current "consumers" of the code path you are proposing to
change and consider whether the performance of any of them may be negatively
impacted by the proposed change. List all these consequences as possible
drawbacks.

## Rationale and Alternatives

This section is extremely important. See the
[README](README.md#rfc-process) file for details.

- Why is this design the best in the space of possible designs?
- What other designs have been considered and what is the rationale for not
  choosing them?
- What is the impact of not doing this?

## Unresolved questions

- What parts of the design do you expect to resolve through the RFC
  process before this gets merged?
- What parts of the design do you expect to resolve through the
  implementation of this feature before stabilization?
- What related issues do you consider out of scope for this RFC that
  could be addressed in the future independently of the solution that
  comes out of this RFC?

---

### Rfcs 20190312 Rpk (`docs/rfcs/20190312_rpk.md`)

- Feature Name: Red Panda Keeper
- Status: draft
- Start Date: 2018-02-09
- Authors:
- RFC PR: (PR # after acceptance of initial draft)
- Issue: (one or more # from the issue tracker)

# Summary

Red Panda Keeper is a binary that will be a Swiss Army Knife CLI application to interact with Red Panda queue. 

In order to provide high productivity and be distinguishable, installation, configuration and operation of the system must be clear, easy and user friendly. That is why Red Panda needs state of the art ops tool. In order to achieve that, it is planned to use modern CLI framework that will handle multiple configuration sources and deliver great user experience. The CLI system will help users and developers to interact with Red Panda in structured, understandable and easy way. 

# Motivation

The RPK is intedented to be the single application used to interact with Red Panda low latency queue. 

It will allow operators to:

- Download & install chosen version of Red Panda
- In future Setup developer cluser in docker-compose (same as Cockroach DB does)
- Tune the system
- Configure the queue
- Interact with the cluster
- Read the WAL files

# Guide-level explanation
TBD
# Reference-level explanation
The Red Panda Keeper CLI is going to be developed in Go using Cobra/Viper CLI development framework. The framework provides easy way to build CLI applications with support for flags, multiple configuration sources and user friendly output. 
## Detailed design
TBD
## Drawbacks
TBD
## Rationale and Alternatives
TBD
### Alternatives
- Oclif - Node Js based CLI framework based on NodeJS
- Knack - Python based CLI framework

## Unresolved questions

---

### Rfcs 20190926 Cluster Controller (`docs/rfcs/20190926_cluster_controller.md`)

- Feature Name: cluster::controller 2-phase local node bootstrap
- Status: draf
- Start Date: 2019-09-26
- Authors: alex
- Issue: (one or more # from the issue tracker)

# Summary

two-phase local node bootstrap.

controller is our global partition. It is a model::ntp("redpanda", "controller", 0) alias.
controller is effectively a global metadata cache. It is a compacted topic.

Upon local node start, restart, crash, etc,
we bootstrap by read the controller contents entirely. We expect the
ammount of data to be actually small, i.e.: less than 1GB of content.

While we are reading the contents, we delegate to
`sharded<cluster::partition_manager> pm` the recovery of particular `ntp` via
`cluster::partition_manager>::manage(model::ntp, raft::group_id)`.
This partition manager holds one `storage::log_manager` instance. The
`storage::log_manager::manage(ntp)` will perform the filesystem recovery of
the log segments for a given `ntp`. Note that a `ntp` may have thousands
of individual log segments.

This proposal removes the knowledge from the storage about which core `ntp`
belong to. It pushes this to the place of knowledge `cluster::controller`



```
              +               Kafka +-----------+
              |                 +               |
              |                 |               |
              |          (a)    |               |
              |                 |               |
              |                 |               |  (h)
              |                 |               |
              |                 v               |
              |               cluster           |
              |                 +               |
              |                 |               |
              |          (b)    |               |
              |                 |               |
              |                 v               | read-path [h]
              |           consensus::raft       |
write-path    |            +          +         |
 [ a-g ]      |       (d)  |       (e)|         |
              |            v          v         |
              |           RPC       storage<----+
              |            +
              |            |        (leader / replica 1)
              |            |
              |            |
              |            |
              |     (f)    |    (g)
              |     +------v-------+
              |     |              |
              |     v              v
              +  replica-2      replica-3

```

# Motivation

We need to have a sound bootstraping and log recovery mechanism
upon a machine restart (from shutdown, reboot, crash, etc)

# Guide-level explanation

`cluster::controller` reads the full `cluster::partition`. It will scan for
all the assignments for `model::node_id` belongs to the machine.

It will build 2 indexes. `model::ntp -> seastar::shard_id` and a
`raft::group_id -> seastar::shard_id`. This abstraction is captured by
the `cluster::shard_table`.

`cluster::controller` _copies_ the full assignment to all cores.
Once `cluster::shard_table` is built (and copied to all cores).

We will do a concurrent recovery of all the `model::ntp`. First,
we delegate the recovery to the `sharded<cluster::partition_manager>`.
Every time we ask the
`cluster::partition_manager>::manage(model::ntp, raft::group_id)` to
manage an ntp and raft group, it will create an entry on the
`storage::log_manager` via the `storage::log_manager::manage(ntp)`.

The `storage::log_manager::manage(ntp)` will perform the particular `ntp`
recovery of all log segments on disk.

Note that this operation is happening concurrently. Currently the
`storage::log_manager::manage(ntp)` uses the `seastar::default_priority`
so it means that recovery is happening fairly across all cores and within
each core doing all the IO for recovery.

# Reference-level explanation

This RFC is attached to an interface prototype.

## Detailed design


1. cluster::controller must perform the bootstrap
2. storage must remove the core-assignment lookup or defer to interface
3. We must wire the protocol through redpanda/main.cc
4. We must include a e2e test for it.

## Drawbacks

Why should we *not* do this?

Recovery now has a constant cost before actual machine bootstrap can occur.
However, as long as controller is small (1GB), two-phase recovery will be a
negligible cost since the _majority_ of the cost will be on recovering the
actual log segments.

In addition, when we hit this bottleneck, we can create a _secondary_ log
with _only_ the assignments for this machine which maybe quite small (thousands)
A cache of assignments.

## Rationale and Alternatives

- Why is this design the best in the space of possible designs?

* First, we have no node bootstrapping yet.
* This design is iterative. It is a simple & correct design for node bootstrap.
* The process can be expanded with time, it just sets up the correct
  expectations for subsystems, namely:
  **  storage does not need to know the cluster assignment at the
      `log_manager` level
  **  Wiring of components happens at `main.cc`


- What other designs have been considered and what is the rationale for not choosing them?

* Static partitioning:
  `model::parition_id::type % smp::core_count` is fragile and offers no real
  load balacing system. For example, core0 will always do a little more work
  in most systems because it is the only core guaranteed to exist.
  Second, having a cluster level knowledge gives us the opportunity to optimize
  physical core placement.
  Third, having a global component allows us to do global optimization
  strategies rather than node-local-maxima strategies

## Unresolved questions


* Wiring through the Kafka API is out of scope.
  Both in the current form and for the Management API

---

### Rfcs 20190928 Record Batch Header Vtype (`docs/rfcs/20190928_record_batch_header_vtype.md`)

- Feature Name: add `v_type` to `model::record_batch_header`
- Status: in-progress
- Start Date: 2019-09-28
- Authors: alex
- RFC PR: (PR # after acceptance of initial draft)
- Issue: (one or more # from the issue tracker)

# Summary

Add an internal type to differentiate between different subsystem batches.

```
/* Detailed source-code truncated for AI context efficiency. */
```

## What is being proposed

First observation

The detection whether or not the record is internal should actually
happen at the batch level. The reason is `record_batch` is our atomic unit for
reads and writes. 

Thus, the next modifications follow:

1. add a `model::record_batch_type` _above_ the CRC line
2. de-virtualize `raft::entry` type

## Why (short reason)

We want other subsystems `raft` and `cluster` in particular to take advantange
of our `storage::log` and related infrastructure to persist changes to disk.
It allows us to have one code base for persisting records and reading them from
disk.

## How (short plan)

We will be extending the `model::record_batch_header` with a
`model::record_batch_type` which is just a wrapper around a `uint8_t`

By default `0` will be used to detect errors while either parsing data from the
Kafka layer or from the RPC layer. 

## Impact

* The wire format will increase by 1 byte on disk
* The Kafka produce API will have to set this byte explicitly to `1`
* Raft configurations will have to set this byte to `2`
* Kafka fetch API will have to skip all records that are not equal to `1`

# Motivation

We want to leverage the existing infrastructure for persisting and reading data
from disk.

# Guide-level explanation

Largely, each new component that wishes to extend the type, should reserve their
integer in the `model` submodule to ensure no collisions at runtime. 

Note that we `explicitly` chose an integer to have no inter-module coupling.

Reading and writing will have no modification other than reading+writing 
the extra byte which should be a very small change. 

The main dependency is that the `kafka::produce` API will have to set the 
byte. 

# Reference-level explanation

Changing the `model::record_batch_header` is a very low level feature. It is 
modifying the disk layout and therefore the readers. We expect that most 
features will never actually extend this.

1. Storage
2. Raft
3. Cluster
4. Kafka

We expect most features will actually interact with modules at the interface 
level and not at the disk format level. That is, you will wrap your _puts_ 
as a Kafka produce API format, or you will interact w/ the global cluster
via the `cluster::controller` module and so on.

The main note worth highlighting is that by setting the default to 0, 
it allows us to detect if the proper subsystems actually set the field 
or simply left it alone. Note that leaving it alone is a bug. It _must_ be
set by each subsystem. 

We expect to have tooling integrated with this feature so system administrators
can actually inspect and see if we have a bug by not having set the 
`model::record_batch_type` in the actual on-disk format.

## Drawbacks

Why should we *not* do this?

* Adds complexity to the parsing protocol
* Adds complexity to every module by allowing every module to set the 
  `model::record_batch_type`
  
## Rationale and Alternatives

- Why is this design the best in the space of possible designs?

  This proposal allows us to leverage the same on-disk format as the 
  `storage::log` _without_ the need for keeping secondary metadata.
  In other words, if we were to _not_ use this design, we would have to 
  have _some_ way of knowing whether the batch on disk is publicly visible
  to the Kafka layer or not. 
  
  The complexities of maintaining alternative metadata is a possible source
  of inconsistencies and a _major_ source of complexity. 

- What other designs have been considered and what is the rationale 
  for not choosing them?

  Setting  a single bit in the `model::record_batch_attributes` was 
  considered and discarded:
  
  -  It is actually taken into account with the Kafka CRC
     Which meant that we would have to re-checksum all payloads
  -  We would have no way of detecting `unset` bits. This is a powerful 
     property of this design. By requiring _each_ module to set this byte
     we can detect if a new extension to the code base simply forgot to 
     set the `model::record_batch_type` and inform early during testing.
     This property could not have been achieved with a single bit.

## Unresolved questions

-  Missing integration with tooling. What and which tool should
   work w/ the log replay functions to help us debug bad log segments

---

### Rfcs 20191003 Rpc Client Timeouts (`docs/rfcs/20191003_rpc_client_timeouts.md`)

- Feature Name: Timeouts in RPC client
- Status: draft
- Start Date: 2019-10-03
- Authors: michal
- Issue: #137

# Summary

The RPC stack is asynchronous (i.e. clients do not block while waiting for the
response). The RPC client implementation uses the correlation map
to identify and dispatch responses for particular requests.
The RPC client returns the `seastar::future<..>` that will fire up continuations
whenever the response is ready. Currently if the response is not delivered
to the client, the returned future will never be completed. In order to prevent
this situations the timeouts have to be introduced into RPC client stack.
The user should be able to pass the timeout as one of the method arguments,
so that each request can have different timeout. When the time of
processing client request is longer then requested timeout the client should
return `make_exception_future<>()` with designated type of exception.

# Motivation

Timeouts are required in RPC stack as waiting for the response
from remote server can last indefinitely.

# Guide-level explanation

## Client API

The timeout parameter will be added as a last optional argument to generated,
client methods. By default there will be no timeout.

```c++
future<ret_t> method_1(agr1_t arg1, rpc::clock_type::time_point timeout)
```

where:

```c++
static constexpr rpc_clock::time_point no_timeout = rpc_clock::time_point::max();
  ```

In order to use an absolute duration as a timeout (f.e. 5 seconds) the caller
have to add `rpc::duration_type` to `rpc::clock_type::now()` every time the method is
called in order to make it timeout after requested amount of time.

Every time the timeout value passed to the method is different than `rpc::no_timeout`
the `rpc::client` logic will arm a `seastar::timer`, before sending
an actual request to server, to fire at requested time point.

When the timer fires it will fail the outstanding `future` related with
particular request so that the `rpc::timeout_exception` will be propagated
to the caller.

## Detailed design

Right now the outstanding requests responses are kept in
`std::unordered_map<uint32_t, promise_t>` where
`promise_t = promise<std::unique_ptr<streaming_context>>`.
The `promise_t` is used to signal the client that the response is ready.
If the timeout has elapsed the promise must be completed with an exception
and removed from the correlation map, additionally all the processing related
with this request should be aborted. In order to do this each request have to
have a timer related with it. The solution is to keep wrap the promise_t with an
additional structure that will keep the timeout timer, only in case when the
timeout parameter was provided. The structure will look like this:

```c++

CONCEPT(
    template<typename R> concept ResponseHandler = requires(R r, std::exception& e) {
    { r.complete_exceptionally(e) } -> void;
    { r.complete(e) } -> void;
    { r.get_future()} -> future<std::unique_ptr<streaming_context>>;
};)

template<typename Impl>
CONCEPT(requires ResponseHandler<Impl>)
struct response_handler {
    template<typename... Args>
    response_handler(Args... args) {
        std::make_unique<Impl>(std::forward<Args>(args)...);
    }

    using response_ptr = std::unique_ptr<streaming_context>;

    future<response_ptr>  get_future() {
        return _impl->get_future();
    }

    void complete_exceptionally(std::exception& e) {
        _impl->complete_exceptionally(e);
    }

    void complete(response_ptr r) {
        _impl->complete(std::move(r));
    }

private:
    std::unique_ptr<Impl> _impl;
};
```

The `Impl` without timeout will only contain a `promise_t` instance whereas
the timeout version will contain a timeout timer of type `rpc::timer_type`.
The timer will be armed during construction of the `timeout_response_handler`
and it will execute the following action (passed as lambda to constructor):

```c++
    auto it = _correlations.find(correlation_id);
    auto response_handler = std::move(it->second);
    _correlations.erase(it);
    response_handler.complete_exceptionally(std::move(ctx));
```

Additionally the `rcp::client::dispatch` method can not fail when the response
with some correlation id was not found in the map.

The timer will be deactivated after when
either `complete()` or `complete_exceptionally()` method will be called.

## Drawbacks

Increased memory usage especially in cases where there are a lot of outstanding
RPC requests.

## Rationale and Alternatives

- Why is this design the best in the space of possible designs?

The design is based of a battle tested Seastar internals.
The Seastar timers implementation is lightweight and it has tunable accuracy.
In the proposed solution the single timeout accounts for all the parts of RPC stack,
serializing and sending request, waiting for response,
therefore the design and implementation is simple and straight forward.

- What other designs have been considered and what is the rationale for not
  choosing them?

Implementing the timeouts based on the polling,
iterating over futures waiting for response every certain amount of time
and failing these futures thats timeout elapsed. The solution would be however
more complicated.

## Unresolved questions

- How to differentiate the situation when server will respond with corrupted
  correlation id from the one when timeout occurred.

---

### Rfcs 20191020 Partition Allocator (`docs/rfcs/20191020_partition_allocator.md`)

- Feature Name: add cluster::parition_allocator
- Status: in-progress
- Start Date: 2019-10-20
- Authors: Alex
- Issue: #223

# Summary

We introduce a new type `cluster::partition_allocator`.
It is a round-robin selector of machines followed by a greedy 
assignment of lowest loaded core.

The proposed algorithm handles machine diversity and rollbacks of
partial assignments (see tests). That is, in the case
that we cannot allocate all replicas on different machines,
we rollback the partial assignments and return an allocation failure.

## What

Adding a centralized place for constraint solving, partition
assignment.

## Why (short reason)

We need a centralized allocator capable of taking  advantage 
of non-uniform hardware. Hardware will
become non-uniform even if it starts out uniform as components fail
throughout the life of the program. Machine and data center hardware
evolution ensure that most of the time, in the long term, we'll
be facing with heterogeneous hardware.

## How (short plan)

We'll be extending the `cluster::controller` to keep track of
machine registrations

## Impact

This is required for v1. Fundamental to product.

# Motivation

We need a centralized place for constraint solving, taking into account
disks sizes, networking, core count, etc.

We expect an even of load given a set of constraints. This
proposal only covers even distribution among cores.

# Guide-level explanation

We introduce a new type `cluster::partition_allocator`. The type is responsible
for even distribution of load across _all cores_ in the cluster.

In addition, it sheds a _little bit_ of load from core 0, by increasing the weight
for the core by _exactly_ 2 partitions, see `cluster::allocation_node::core0_extra_weight`.

The idea is that core-0 is already busy with
default-allocated work for all the `cluster::controller*` logic

# Reference-level explanation

`cluster::controller` needs to delegate the actual _physical_ placement
of partitions to a scheduler. The oracle should solve 2 problems:

1. even distribution of load across _all_ cluster cores
2. do partition allocation respecting constraints (taints & tolerations)
   Note: We do not discuss taints and tolerations in this design
   as we are postponing the discussion and implementation of
   advanced soft/hard constraints, commonly known as taints & tolerations[1]

The flow of information is

Kafka Admin -> _local_ controller -> _global_ controller -> partition_allocator


## Telemetry and Observability

* The telemetry is observed locally at each node with the number
  of partitions assigned to each node

* Partition alloation telemetry is deferred

## Detailed design - how


Assume this POD

```
struct allocation_node {
    /// physical machine
    model::broker _node;
   
    /// number of partition allocations left
    /// initialized to (#cores * 7000) - 2
    uint32_t _partition_capacity;
   
    /// each index is a CPU. A weight is roughly the number of assigments
    /// _weights.size() == # of cores
    std::vector<uint32_t> _weights;
};

```


The general algorithm for balancing _# of cores_ only:

* Compose a list of all nodes with capacity to allocate partitions
** This is kept in an intrusive linked list, when nodes become full
   the list removes the node from the available nodes with capacity
   
* iterate over all partitions
* for all replicas, choose next computer with our round robin pointer

** if no match, rollback _all_ previous allocations and return std::nullopt 



At the node level
`allocation_node::allocate()` gets us the lowest loaded core via
`min_element()` (by weight), which ensures perfect balance within the machine.

```
    uint32_t allocate() {
        auto it = std::min_element(_weights.begin(), _weights.end());
        *it++; // increase weight
        return std::distance(_weights.begin(), it);
    }

```

This algorithm gives us round-robin assignment for diversity of machines 
and good balance per _individual machine_ at the core level.


## Drawbacks

## Rationale and Alternatives


### Why is this design the best in the space of possible designs?

This design is very simple. We keep a single pointer to the
the next element to do round robin, and when that element is 
saturated, we remove it from the list. We attempt to add it on
deallocation if we later remove the assignment.

The solution however ignores rack diversity, soft constraints
also known as affinity, anti-affinity for negative
constraints as well as _hard constraints_ whic make
the affinities a failure on allocation.

These techniques require a much more sophisticated scheduler,
similar to the `max-min`[1] proposed in the Apache Mesos paper
which when taken multiple resources into account is known as
Dominant Resource Fairness (DRF)[2]


### What other designs have been considered and what is the rationale for not choosing them?

- DRF - It is correct, decays correctly into max-min
  and used at Google for Borg, Kubernetes and Apache Mesos to run
  very large workloads.

  DRF is complicated and requires more unification of models & constraints.
  It is unreasonable to expect a good implementation in 1 week of DRF.
  
- std::shuffle() + random node selection. The shuffle as the list of nodes increases
  is very expensive. It calls the move ctor on all elements effectively.

- jump consistent hash + linear probing proved required a lot of state to keep track
  of which CPU's have been assigned to make the correct jump. The implementation
  was much more complex with very little gain over std::shuffle
  
- rendezvous hashing should also provide a decent balance

```
Destination = namedtuple('Destination', ['host', 'hash', 'weight'])


def score(hash_value, weight):
    return -weight / math.log(hash_value / HASH_MAX)


def pick_destinations(key, destinations, k=1):
    key_hash = hash_key(key)
    annotated = [(score(merge_hashes(key_hash, dest.hash), dest.weight), dest.host)
                 for dest in destinations]
    ordered = sorted(annotated)
    return [host for _, host in ordered[:k]]

```

- Ceph CRUSH algorithm is also an option for load balancing

### What is the impact of not doing this?

N/A. We need _some_ partition allocator for the product.

## Unresolved questions


* Internal RPC discussion in case of controller forwarding
* usage of `machine_lables` is deferred discussion for RPC RFC
* Taints & Tolerations scheduling, also known as soft and hard
  constraints.
* Interface exposed to the Kafka Admin API
* Prometheus Telemetry & Observability is deferred,
  covered in Issue #224
* We don't yet take into account system saturation.
  We throw an exception when we've assigned 7K partitions to
  a particular core which should be about 7TB on the fastest collection of segments (1GB)
  and about 112TB on a 16core machine.


## References

[1] https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.16/#pod-v1-core
[2] max-min: https://en.wikipedia.org/wiki/Max-min_fairness~
[3] DRF: https://cs.stanford.edu/~matei/papers/2011/nsdi_drf.pdf

---

### Rfcs 20191023 Feature Flags Modes (`docs/rfcs/20191023_feature_flags_modes.md`)

- Feature Name: Redpanda modes and feature flags
- Status: in-progress
- Start Date: 2019-10-23
- Authors: David Castillo
- Issue: [#226](https://github.com/redpanda-data/v/issues/226), [#227](https://github.com/redpanda-data/v/issue7/227)


# Summary

Sometimes, settings, checks and tuners that are desirable for a production environment
aren't necessary for someone who's just evaluating redpanda on their personal laptop
or on a small machine. These settings can even crash the developer's computer
(see [#226](https://github.com/redpanda-data/v/issues/226)), which is not a good first
impression.
To avoid these issues, we should have feature flags to be able to control some
redpanda settings (described below) in a more fine-grained scope, while also making
it easy for the user to run redpanda without having to go over each option.
In a nutshell, we should have a simple way to allow developers to disable production
settings when they're running redpanda in their personal computers or in an evaluation
environment.

## What is being proposed
We introduce the concept of feature flags and execution modes to redpanda. We want to
allow the user to enable and apply the tuners and optional redpanda settings that they wish.

At the moment of writing this proposal, these are:

**Tuners**:
Net
- IRQ balance banning
- IRQ affinity
- Receive Packet Steering (RPS)
- Receive Flow Steering (RFS)
- NTuple tuning
- XPS queues setup
- RFS table setup
- Listen backlog size tuning
- SYN backlog size tuning
Disk
- IRQ balance banning
- IRQ affinity
- No-op scheduling
- NoMerges
CPU
- HT tuner disabling
- CPU governor setup
- Power saving modes disabling (requires to setup GRUB and reboot machine**

**Runtime features**
- Memory locking
- Real time scheduling

**Optional configuration files**
- Custom coredump pattern & coredump processing script.

Currently, redpanda's DEB and RPM packages also ship with a custom coredump processing
script, enabled through a sysctl
[.conf file](https://www.freedesktop.org/software/systemd/man/sysctl.d.html#), which is
automatically installed with the package.
While this script is useful if the user wants to save a coredump for later analysis, it
puts a very high load on IO for common development environments, freezing or crashing
them due to the size of the core dump, which is usually over 50GB. Because of this, the
installation of this script should be a choice and not enabled by default, so a feature
flag should be created for it too. In general, all optional system configuration files
should have a feature flag to allow rpk or redpanda to install them if the user chooses
to. The use case described above is the main reason for this proposal.

Additionally, we introduce "development modes", which are a way to easily enable or
disable groups of features at once. Specifically, we propose the implementation of
2 main modes:

- **development**, which removes the custom coredump pattern configuration file, and the
  processing script it enables; an disables tuners that might reduce the overall perceived
  performance of a development environment, which usually has limited resources shared
  among many heterogeneous processes.
- **production**, which enables all the tuners, as well as some seastar features such
  as memory locking and real-time scheduling.

## Why (short reason)

To allow the developers working on, evaluating, or locally developing software
that depends on redpanda, to disable settings and features that might make the
experience of using redpanda bad or confusing.

## How (short plan)

- Adding an `rpk` section to the `redpanda.yaml` config file, where each tuner can be
  enabled (set to `true`) or disabled (set to `false`).
- Exposing the relevant seastar configurations, such as memory locking and
  real time scheduling, in the `redpanda.yaml` configuration file under the `rpk`
  section.

These configuration fields would serve as feature flags for redpanda's tuners and
settings, as well as seastar's.

To save the users the unnecessary effort of disabling or enabling these settings
themselves, a new rpk command is proposed, called `mode`. When executed (i.e. `rpk
mode development`), this command will set the relevant fields to `false`, so that the
user can get panda up and running in their environment, without affecting other processes.

## Impact

This would greatly improve the developer and user experience, so it's really
important for the first release.

# Motivation

Currently some redpanda and seastar settings, although meant to make redpanda's
performance as good as possible in a production setting, make other processes halt
or slow down, and could even crash the developer's machine. For example, the
redpanda deb and rpm packages install a custom systemd coredump script that saves all
coredumps in `/var/lib/redpanda/coredump`. If redpanda crashes, the core dump data
can take more than 50 GB, which can bring their computer to a halt. Because the
redpanda systemd service is enabled by default and is set to restart after crashing,
if the user should reboot their computer this process could go on and on filling the
user's storage.

Likewise, seastar's real time scheduling may decrease the performance of other processes
usually present in developers' machines, such as IDEs, music streaming apps, and even
shells.

Because of this, we should allow users and developers to disable these settings in a
simple way.

# Guide-level explanation

To achieve the best performance possible, redpanda relies on interrupt coalescing, interrupt
distribution, no-op IO scheduling, CPU frequency scaling (performance governor), NIC
multi-queue setup, memory locking and real-time scheduling. However, enabling these features
isn't always the best for a personal computer running multiple processes, each of which are
always competing for memory, CPU and IO.

Because of this, redpanda enables them by default, but allows them to be disabled according to
your needs. To do it easily, you can toggle on redpanda's Developer Mode, which disables these
production settings so that you can still use it in your laptop.

Enabling Developer Mode is as simple as running

```sh
# You may need to run this command as root.
rpk mode developer
```

For runtime settings (that is, settings that are only effective while redpanda is running),
it will conveniently disable any which may affect other processes' performance, by setting the
corresponding fields in the `redpanda.yaml` configuration file to `false`.

For persistent session or system configurations, it will also undo any changes redpanda might
have made in your machine (such as enabling real-time scheduling or disabling certain interrupt
requests for some CPUs).

Likewise, if you'd like to re-enable the default, production settings, just run

```sh
# You may need to run this command as root.
rpk mode production
```

If you'd like to review everything `rpk mode` will do, you can also run it interactively with
the --interactive flag:

```sh
# You may need to run this command as root.
rpk mode developer --interactive
```

Some redpanda tuners change GRUB configurations, thus requiring a restart for them to take
place. You may choose to disable those by adding `--with-reboot-options false` (or enable
them, passing `true`):

```sh
# You may need to run this command as root.
rpk mode production --with-reboot-options false
```

These commands are idempotent, so they're safe to run multiple times.

For runtime settings, you can also review and edit `redpanda.yaml` and enable or disable
any of them. However, if you run `rpk mode` again, it might overwrite any changes you
made in the affected fields.

If you switch modes or change any setting, you have to restart redpanda for the changes
to take place.

# Reference-level explanation

## Detailed design

The user-facing changes, as mentioned in the previous section, include:

1. Adding a new command (`mode`) to rpk:

  The `mode` command should disable the redpanda tuners and features - by setting their
  respective fields in the `rpk` section in `redpanda.yaml`-, as well disabling OS features
  and settings - by running any commands and editing any system files necessary -, that
  might not play well with the multitude of processes usually active in developer machines.
  
  Editing `redpanda.yaml` should be done by loading the current contents, changing the
  necessary values only, and writing the file again, so as not to unnecessarily overwrite
  any unrelated config fields, such as a port or an IP address.
  
  For running processes the current `Proc` interface available in rpk can be used.

2. Adding an `rpk` section in the config file.
  
  The new section

  ```yaml
  ---
   rpk:
     network:
       ban_irq_balancing:    true | false
       irq_affinity:         true | false
       rps:                  true | false
       rfs:                  true | false
       rfs_table:            true | false
       ntuple:               true | false
       xps_queues:           true | false
       listen_backlog_size:  true | false
       syn_backlog_size:     true | false
     
     memory:
       memory_locking:       true | false

     cpu:
       real_time_scheduling: true | false
       disable_ht_tuner:     true | false
       cpu_governor:         true | false
       disable_power_saving: true | false
 
     disk:
       ban_irq_balance:      true | false
       irq_affinity:         true | false
       noop_scheduling:      true | false
       nomerges:
       
     coredump:
       install_custom:       true | false
  ---
  redpanda:
    # current redpanda settings 
   
  ```
  
  should be included in the `Config` struct in rpk, with the appropriate tags for the
  yaml parser, like so:
  
  ```go
  type Rpk struct {
      Network       `yaml:"network"`
      Memory        `yaml:"memory"`
      CPU           `yaml:"cpu"`
      Disk          `yaml:"disk"`
      Coredump      `yaml:"coredump"`
  }
  
  type Config struct {
	  Rpk       `yaml:"rpk"`
      Redpanda  `yaml:"redpanda"`
	  // current fields...
  }
  ``` 
  
  The values can then be queried by any part of the rpk code to enable or disable features:
  
  ```go
  if config.Rpk.realTimeScheduling {
      err = os.NewProc().RunWithSystemLdPath("sysctl", "-w", "kernel.sched_rt_runtime_us=-1")
  }
  ```
  
  Some of the fields in the new config section which correspond to runtime settings will
  not translate to specific commands, but will be passed to redpanda/ seastar as flags.

## Drawbacks

Arguably, this adds complexity to redpanda/ rpk. If a way is found to run redpanda safely
on a development environment, that doesn't require exposing a choice to the user, it
should be preferred over this one. For example, detecting the available resources and
deciding whether to activate or deactivate a feature or tuner.

On a related note, some users **will** skip the docs and run redpanda/ rpk right
away, so this doesn't take away the underlying risks mentioned above.
Likewise, it also opens a way to run redpanda in a way that limits its performance.
For example, a user could install redpanda on a production machine and run
`rpk mode developer`, forgetting to change it back, leading to inferior performance.

## Rationale and Alternatives

Having fine-grained configuration could greatly simplify setting up redpanda in a developer
machine.
Also, having development and production modes makes it super easy for a user to run redpanda
without having to familiarize themselves with the different OS settings redpanda depends on,
and then judging whether they should enable or disable it, or researching which command they
should run to turn them on or off.

Additionally, its implementation is trivial:
- Adding new commands is pretty straightforward with cobra (the cli lib used in rpk)
- New config fields need to be added to the yaml file, and loaded into the `Config`
  struct, which is done with go's field tags and the go-yaml library.
- The most involved part is identifying which checks, tuners and seastar flags should be
  disabled with the development mode, and how to undo production mode changes made to the
  machine if the user should decide that they want to run in development mode after all
  (see the unresolved questions section).

### Alternatives considered

1. Having 2 separate **services**, one which doesn't allow further configuration, and a
  [template service](https://fedoramagazine.org/systemd-template-unit-files/) that allows
  the user to override the default (production) config passing flags to the `ExecStart`
  command (e.g. `systemctl start redpanda@--development-mode.service`).
  This alternative, although it makes it straightforward to start redpanda in development
  mode with systemd, requires us to have 2 different service files, which can be
  [inconvenient for some users](https://github.com/redpanda-data/v/issues/212#issuecomment-544196619),
  and doesn't allow for installing/ uninstalling files (like the default coredump script),
  which `rpk mode` does.
  
2. Having 2 different **packages**, one for dev mode, which doesn't install the systemd
  coredump script, and a production one that runs redpanda with `--lock-memory true` and the
  coredump script. This alternative could be very confusing for users, and adds a lot of
  overhead to our current packaging and installation process, since we'd have to maintain 2
  packages with minimal - albeit fundamental - differences.
  
3. Detecting which features it makes sense to activate, and which tuners rpk should run,
  based on the environment's resources at the moment of execution. This would be a 100%
  hands-free experience for the user, but it's a more involved solution that could take
  more than the time we have before redpanda's first release deadline (1 week at the time
  of writing). It should however be re-evaluated in the near future since it could greatly
  improve the user experience.

4. Having only one new field, `developer_mode`, and a new flag `--developer-mode` for rpk's
  `tune`, `start`, and `check` commands. This alternative was discarded in favor of more
  fine-grained control over which features and tuners should be toggled on or off.

## Unresolved questions

Should redpanda store the current machine's relevant state before switching to developer mode, so it
can reliably roll the changes back?

---

### Rfcs 20191122 Precalculated Iotune Info (`docs/rfcs/20191122_precalculated_iotune_info.md`)

- Feature Name: Precompiled iotune info
- Status: draft
- Start Date: 2019-11-22
- Authors: David Castillo <david@vectorized.io>
- Issue: #200

# Executive Summary

One of the tools rpk relies on for tuning the machine where redpanda will run is
[iotune](https://github.com/scylladb/seastar/tree/master/apps/iotune). It runs
IO benchmarks to find the optimal read/write IOPS and IO bandwidth values, which
are used to configure the Seastar IO scheduler. To get the best results, iotune
should be run for at least a couple of minutes, up to ~45 minutes. This means
that even if the redpanda installation, tuner execution and startup can take as
few as 1 minute, the iotune step can make the whole process seem like an
eternity for our users.

However, it is expected that many of Vectorized's clients will run redpanda in
any of the most popular cloud vendors' (AWS, Azure, GCP) standard VM types. We
can run iotune on the VM/ elastic storage volume types that we project will be
the most used for redpanda, and encode the obtained data into rpk to decrease
further the time it takes for an user to have redpanda up and running.

## What is being proposed

a. To include precompiled iotune results from all the recommended major cloud
vendor/ VM/ storage type combinations, and including them in rpk.

b. Additionally, to collect VM metrics such as CPU type, CPU Features, disk
type, RAID setup, network settings, memory settings, NUMA socket settings,
hyperthread settings, kernel version settings, operative system settings.

## Why (short reason)

a. To reduce the time it takes to set up redpanda and run it, and to make the best
first impression redpanda can as a product.

b. To be able to map the resulting iotune values to not only a VM type, but more
generally to OS and hardware features and settings.

## How (short plan)

1. Creating a Python script to iterate over a major cloud vendor/ VM type/
storage device type matrix, leveraging our
[Terraform configuration](https://github.com/redpanda-data/v/tree/master/infra)
to deploy test nodes, overriding its default behavior, which is to run the rpk
tuners and then redpanda, with a script to
2. Running iotune on the deployed VMs,
  2.1. Collecting the VM OS settings and hardware features described above
3. Uploading each result to s3 for analysis and aggregation.
4. Aggregating the collected data to embed it into rpk as a vendor to
machine-type to storage-device-type matrix, which will enable the user to skip
iotune execution or the related configuration, and start redpanda right away.

Supported cloud vendors:
- AWS
- GCP
- Azure

Recommended VM types by vendor:

**AWS**
> See https://aws.amazon.com/ec2/instance-types/ for detailed info on instance
> type specifications
> EBS-only instance types aren't included because networked IO is throttled
- Compute optimized
  - c5d: large, xlarge, 2xlarge, 4xlarge, 9xlarge, 12xlarge, 18xlarge,
    24xlarge, metal
- Storage optimized
  - i3: large, xlarge, 2xlarge, 4xlarge, 8xlarge, 16xlarge, metal
  - i3en: large, xlarge, 2xlarge, 3xlarge, 6xlarge, 12xlarge, 24xlarge, metal

**GCP**
> See https://cloud.google.com/compute/docs/machine-types for more info on GCP
> machine types
- General purpose
  - n1: standard, highcpu (in their 2, 4, 8, 16, 32, 64 and 96 vCPUs variants)
  - n2: standard, highcpu (2, 4, 8, 16, 32, 48, 64, 80 vCPUs variants)
- Compute optimized
  - c2: standard (4, 8, 16, 30, 60 vCPUs variants)

All will be tested with Local SSDs (SCSI and NVMe). Details on GCP block
storage types at
https://cloud.google.com/compute/docs/disks/performance.

**Azure**
> See the virtual machine specs at
> https://docs.microsoft.com/en-us/azure/virtual-machines/linux/sizes
- General purpose
  - Dsv3: Standard\_D{2, 4, 8, 16, 32, 48, 64}s\_v3
  - Dasv4: Standard\_D{2, 4, 8, 16, 32, 48, 64}as\_v4
  - Dv3: Standard\_D{2, 4, 8, 16, 32, 48, 64}\_v3
  - Dav4: Standard\_D{2, 4, 8, 16, 32, 48, 64}2a\_v4
  - DSv2: Standard\_DS{2, 3, 4, 5}\_v2
  - Dv2: Standard\_D{2, 3, 4, 5}\_v2
  - Av2: Standard\_A{2, 4, 8}\_v2, Standard\_A{2, 4, 8}m\_v2
- Compute optimized
  - Fsv2: Standard_\F{2, 4, 8, 16, 32, 48, 64, 72}s\_v2
- Storage optimized
  - Lsv2: Standard_\L{8, 16, 32, 48, 64, 80}s\_v2
  
All will be tested with
[Standard, Premium and Ultra SSDs](https://docs.microsoft.com/en-us/azure/virtual-machines/windows/disks-types),
in all their sizes.

## Impact

This will reduce the amount it takes to get started with redpanda, leading to
a better user experience.

# Motivation
## Why are we doing this?

The time available for a product to create a good first impression on the user
(what we refer to as time-to-wow, or T2W) is very little. We hypothesize that
this is specially true for software experts, which have probably heard that some
new tool is "the next silver bullet" one-too-many times.

Because of this, we have a goal of keeping the T2W at 60 seconds or less (T2W
budget).
Having a precompiled data set for the optimal settings for the recommended VM/
storage setups will allow us to reduce the time it takes for a user to be
convinced that redpanda and Vectorized will deliver what we promise.
Essentially, we're turning something that takes 45 minutes into millisecond
scale.

## What use cases does it support?

Ease of installation, ensuring the best performance.

## What is the expected outcome?

A redpanda node, working with near-optimal settings within our T2W budget.

# Guide-level explanation
## How do we teach this?

`rpk` comes with a command called `iotune` which finds the optimal configuration
for redpanda's IO scheduler, with the current storage device and processor. To
get the most precise results, iotune should usually run for approximately 30
minutes - or 1800 seconds (i.e. `rpk iotune --duration 1800`).

However, rpk ships with a matrix of optimal configurations for the
recommended VM/ storage device type setups on AWS, GCP and Azure. Thanks to
this, when running redpanda on the recommended setups, you can just fire it up
with no additional steps.

> It is still recommended to execute iotune on custom setups. There's no need
> to run it every time, however: you can run it once for each setup that you
> will use to run a redpanda node, keep the results, and reuse them from there.

Upon statup, rpk will try to detect the current cloud and instance type via the
different vendors' metadata APIs, setting the correct iotune properties if the
detected setup is a supported one.

If access to the metadata API isn't allowed from the instance, you can also hint
the desired setup by passing the `--well-known-io` flag to `rpk start` with the
cloud vendor, VM type and storage type surrounded by quotes and separated by
colons (`:`):

```sh
rpk start --well-known-io 'aws:l3.xlarge:io1'
```

It can also be specified in the redpanda YAML configuration file, under the
`rpk` object:

```yaml
rpk:
  well_known_io: 'gcp:c2-standard-16:nvme'
```

> If `well-known-io` is specified in the config file, and as a flag, the value
> passed with the flag will take precedence.

In the case where a certain cloud vendor, machine type or storage type isn't
found, or if the metadata isn't available and no hint is given, `rpk` will
print an error pointing out the issue and **continue** using the default values.

# Reference-level explanation

## Interaction with other features

The cloud vendor and VM type detection will be on by default, and on unsupported
VMs or VMs where access to the vendor's metadata API isn't allowed, the vendor
and VM can be hinted `rpk` by passing a new flag, `--well-known-io`. This aligns
with the UI built so far.

The only forseen conflict is with the `--io-properties` flag in the start command,
since they both serve as a way of specifying the source for Seastar's IO
scheduler configuration. If both flags are passed, `rpk` will print a message
prompting the user to pick only one and stop.

Hopefully, `--well-known-io` will make easier for a large percentage of our
users to run redpanda. However, there will be some users which will want to run
redpanda in their own infrastructure or on non-recommended setups. Because of
that, the `iotune` command is still useful and won't be replaced or made
obsolete by this.

## Telemetry & Observability

Described in the sections below.

## Corner cases dissected by example.

As explained in **Interaction with other features**, if the user also passes the
`--io-properties` flag along with `--well-known-io`, or when setting a value for
`rpk.well_known_io` in `redpanda.yaml`, this is considered a conflict that the
user is expected to resolve.

## Detailed design 

### What needs to change to get there

The implementation of this feature will be divided in 3 steps.

1. Use the
  [available iotune result data](https://github.com/scylladb/scylla/blob/master/dist/common/scripts/scylla_io_setup)
  for AWS i3 instances, collected by the ScyllaDB team, and embed it into rpk.
  This could happen even before this RFC is approved, in order to have
  something ready for the scheduled 2019-12-15 release.
2. Deployment and data collection and aggregation, and extension of the initial
  data included in rpk during step 1 with the new data.

### How it works

#### Step 1

To achieve step 1, the `--well-known-io` flag will be added to the `start` command,
following the same patterns used by all flags in `rpk`. This flag's effect will
be to look up the configuration for the given vendor, VM, and storage type - in
an in-memory matrix, and to propagate the settings to redpanda/ seastar via its
[`--io-properties` flag](https://github.com/scylladb/seastar/blob/master/src/core/reactor.cc#L3225).

#### Step 2

This step is the bulk of the feature's development. It will be divided into 3
phases too:

a. Collecting data for AWS setups
b. Collecting data for GCP setups
c. Collecting data for Azure setups

This part is divided to provide incremental updates, and because it is expected
that the Terraform module for each vendor will be different from the other ones.
The order is determined to provide the updates as fast as possible; there's
already a module to deploy on AWS, so the AWS data will be collected first.
Some potential clients use GCP too, so it will come second, and finally, Azure
data will be collected.

When each phase is completed, the results will be integrated into rpk building
upon step 1's progress.

In order to deploy each vendor's module with its different VM-storage type
combinations, a python script will be created to iterate over the vendor/ vm/ 
storage matrix described in the **How (Short Plan)** section. It will leverage
Terraform's action's `--state-out` and `state` flags to be able to use the same
module, but manage multiple simultaneous deployments.

The state file names will have the format
`<vendor>-<vm type>-<storage type>.tfstate`, thus making deployments easily
traceable to their state files.

Each phase will consist of:
- Deploying all the specified VM/ storage type combinations, and choosing for
  either RHEL 8 and Ubuntu 18.04 randomly
- Running iotune
- Saving the iotune data in an S3 bucket
- Repeating this process a couple times to be able to average the results
- Aggregating the results
- Adding the new data to `rpk`

This will be done using Terraform's `remote-exec` provisioner, which allows
executing commands on the deployed machine, and a bash script that it will
provision to each deployed VM, encoding the process above as a bash script.

The structure of the data in S3 will be
```
root-bucket
|- iotune-<date> 
   |- <vendor>
   |   |- <vm type name>
   |      |- <uuid>
   |         |- <storage type name>
   |            |- io-properties.yaml
   |            |- vm-data.txt
   |- <vendor b>
...
```

## Drawbacks

The deployment of hundreds of VMs, many of which belong to the highest tiers,
will no doubt be super expensive. However, it's something that doesn't need to
be done so frequently (only when new VM and storage types are launched), and the
improvement this feature poses in terms of UX is great, reducing the time it
takes to start redpanda for the first time from ~30 minutes to <1 minute, all
while it ensures that the user is running redpanda with the best IO config.

## Rationale and Alternatives

As mentioned before, the feature is fairly easy to implement, but the challenge
is in the deployment and the data collection. However, it will be a proof of how
redpanda will go the last mile for its customers, and that we're always one step
ahead.

Additionally, it will provide us with a very broad picure of the approximate
performance baseline across clouds and storage types.

### Alternatives considered

- Not including the precompiled iotune data
  Keeping `rpk` as is, and deferring the optimal IO settings to run redpanda to 
  the user would certainly be cheaper in terms of effort, as well as
  financially, but its likely that our users will use either AWS, GCP or Azure,
  so important that we make it as easy for them to run redpanda on the cloud.

## Unresolved questions
- What other data would be valuable to collect, besides the dataset described
  above?

---

### Rfcs 20200421 Raft Recovery (`docs/rfcs/20200421_raft_recovery.md`)

- Feature Name: Raft lazy flush
- Status: completed
- Start Date: 2020-04-21
- Authors: Michal
- Issue: ch233

# Executive Summary

This change proposing not flushing during recovery or during non-quorum writes.

## Why (short reason)

In redpanda we define consistency levels. For now Raft differentiate the
`quorum_ack` from `leader_ack` and `no_ack` (both of them are treated the same
way). The `quorum_ack` use classic Raft algorithm to replicate the changes
across the cluster. For `leader_ack` and `no_ack` we return success after
successful append to the leader log and we relay on raft recovery to replicate
data to follower logs. This makes recovery to be first class citizen of our Raft
algorithm implementation which is different to the one used in classic
implementation where recovery is exceptional situation and happen only after
failures. This makes the need to make recovery, being an async replication, fast.

## How

By default classic Raft algorithm implementations rely on the fact that entries
appended to follower logs are non volatile (flushed). Thanks to our log
implementation we can reliably track both the committed
(flushed and nonvolatile offset) and dirty (not flushed offset).
Raft implementation can then delay flushing follower log while still keep
tracking what entries are safe to be committed.

## Impact

Thanks to the proposed solution redpanda raft implementation will efficiently
support both the `quorum_ack` and relaxed consistency models.

# Reference-level explanation

## Current implementation

In current raft algorithm implementation all raft domain indexes are based
on `storage::log::dirty_offset`.

### Replicate algorithm - `consistency_level::quorum_ack`

Stop conditions:

  1. leader commit_index > offset of an entry that was replicated to
     the leader log (SUCCESS)
  2. term has changed (FAILURE)
  3. current node is not the leader (FAILURE)

Algorithm steps:

  1. Append entry to leader log without flush
  2. Dispatch append entries RPC calls to followers and in parallel flush
     entries to leader disk
  3. Wait for (1),(2) or (3)
  4. When
     ->(1) reply with success
     ->(2) or (3) check if entry with given offset and term exists in log
       if entry exists reply with success, reply with false otherwise

                           N1 (Leader)        +
                           +-------+          |
                       +-->| Flush |--------->+     OK
                       |   +-------+          |     +----(1)----> SUCCESS
                       |                      |     |
     N1 (Leader)       |   N2                 |     |
+-------------------+  |   +--------+-------+ |     |
|Replicate |Append  |--+-->+ Append | Flush |-+---->+
+-------------------+  |   +--------+-------+ |     |
                    |  |                      |     |
                    |  |   N3                 |     |
                    |  |   +--------+-------+ |     +-(2)-(3)---> FAILURE
                    |  +-->| Append | Flush |-+     ERR
                    |      +--------+-------+ |
                    |                         +
                    v               Wait for (1) or (2)
        Store entry offset & term

This algorithm indicates that on the followers log is flushed before response is
returned to the leader thanks to that leader can use the returned `last_log_index`
to establish the latest offset that has been safely (persistently) replicated on
 majority of nodes and using this information update the `commit_index`.

### Replicate algorithm - `consistency_level::leader_ack` and `consistency_level::no_ack`

Algorithm steps:

  1. Append entry to leader log without flush
  2. If append was successful reply with (1) success,
     reply with an error (2) otherwise

                           OK
                            +----(1)----> SUCCESS
                            |
     N1 (Leader)            |
+-------------------+       |
|Replicate |Append  |------>+
+-------------------+       |
                            |
                            |
                            +----(2)----> FAILURE
                           ERR

Follower replication is done in async manner as leader log is ahead of the
follower and recovery is triggered.

Flushing strategy:

- Leader log is flushed periodically (currently using vote_timer)
- Follower log is flushed on each append entry coming from recovery

### Recovery algorithm

`match_index` - greatest index where leader entry term
                is equal to follower entry term

1. Start recovery if follower append_entries/heartbeat returned failure
   and follower `match_index` is smaller than leader log `dirty_offset`
2. If follower `last_log_index` is greater than leader `last_log_index`
   establish new `match_index` by sending `append_entries` moving back
   one batch at a time. Follower will reply with success and this reply
   `last_log_index` is new `match_index`
3. Read up to 32KB of record_batches and send them as `append_entries_request`
   to follower until its `match_index` isn't equal to leader last log index

## Problems with current approach

Recovery happens in small batches (either 32KB or one batch, whichever is bigger)
and as the `append_entries` request is sent to the follower current flushing
policy force to it flush the underlying log before responding to the leader.
Follower has to flush the log to achieve correctness as leader will use
returned `last_log_index` to update its `commit_index`.

## Relation between committed and dirty offset

                    Committed batches
             +-----------------------------+
             |                             |
             v                             v
             +-------------------------------------------------+
             |Batch #1 |Batch #2 |Batch #3 |Batch #4 |Batch #5 |
             +-------------------------------------------------+
Offset       0         2         5         7         9         12
                                           ^                   ^
                                           |                   |
                                           +-------------------+
                                               Dirty batches
Committed offset: 7
Dirty offset: 12

Proposed change relies on fact that batches that were already committed
are non volatile while the one that are 'dirty' are appended to the log already
but can be lost in case of failure.

### Raft related observations

- Leader can only commit (update `commit_index`) those entries which are
  flushed on the majority of nodes
- In order to successfully append entries to the log followers
  have to receive entries with offset that follows its log `dirty_offset`

## Proposed solution

The relation described in previous point is a key to proposed solution.
While replicating with `quorum_ack` raft can only respond to caller when entry
was committed by raft protocol in order to minimize latency we want to do it
in one roundtrip round. For relaxed consistency levels we do not
require majority to persistently store replicated entries before
returning to the caller. In order to achieve performance gain while replicating
entries during recovery we can postpone updating `commit_index` on the leader
and allow follower not to flush on every `append_entries` request coming
from the leader. This will speed up sending 32 KB recovery chunks as leader
will not have to wait for follower to flush before sending next chunk. Leader
`commit_index` will eventually be updated only after it will receive information
from follower that it flushed its log up to the certain point. We do not have to
set `flush_after_append` even for `quorum_ack` recovery as heartbeats will
contain followers committed and dirty index so leader can use them to update
followers state.

## Detailed design - What needs to change to get there

### Changes in leader

Leader has to be able to track both the flushed and dirty offsets
from the followers. So it can use the dirty offset to establish the latest
follower index that matches leader in order to trigger recovery if required.
The committed offset on the other hand is used to establish the offset that was
persistently replicate across majority of nodes.
In order to make it possible new fields have to be added to
`follower_index_metadata`:

    ```c++
        struct follower_index_metadata {
            explicit follower_index_metadata(model::node_id node)
            : node_id(node) {}
            model::node_id node_id;
            // index of last known log for this follower
            model::offset last_committed_log_index; // <----------- New field
            // index of last not flushed offset
            model::offset last_dirty_log_index; // <----------- New field
            // index of log for which leader and follower logs matches
            model::offset match_index;
            // Used to establish index persistently replicated by majority
            constexpr model::offset match_committed_index() const {
                return std::min(last_committed_log_index, match_index);
            }
            // next index to send to this follower
            model::offset next_index;
            // timestamp of last append_entries_rpc call
            clock_type::time_point last_sent_append_entries_req_timestamp;
            uint64_t failed_appends{0};
            bool is_learner = false;
            bool is_recovering = false;
            // indicates if leader has active connection to this node
            bool disconnected = false;
        };
    ```
The addition of `last_committed_log_index` and `last_dirty_log_index`
leader can track both flushed and dirty offsets of followers.

### Changes in follower

In order to allow leader to track both committed and dirty offsets have to be
send in `append_entries_reply`. This will be done in both the heartbeat and
`append_entries` containing data.

    ```c++
    struct append_entries_reply {
        enum class status : uint8_t { success, failure, group_unavailable };
        /// \brief callee's node_id; work-around for batched heartbeats
        model::node_id node_id;
        group_id group;
        /// \brief callee's term, for the caller to upate itself
        model::term_id term;
        /// \brief The recipient's last log index after it applied changes to
        /// the log. This is used to speed up finding the correct value for the
        /// nextIndex with a follower that is far behind a leader
        model::offset last_committed_log_index; // <----------- New field
        model::offset last_dirty_log_index; // <----------- New field
        /// \brief did the rpc succeed or not
        status result = status::failure;
    };
    ```

Additionally leader has to be able to control
if follower should flush while processing `append_entries_request`.

    ```c++
    struct append_entries_request {
        using flush_after_append = ss::bool_class<struct flush_after_append_tag>;

        append_entries_request(
        model::node_id i,
        protocol_metadata m,
        model::record_batch_reader r,
        flush_after_append f = flush_after_append::yes) noexcept
        : node_id(i)
        , meta(m)
        , batches(std::move(r))
        , flush(f){};
        ~append_entries_request() noexcept = default;
        append_entries_request(const append_entries_request&) = delete;
        append_entries_request& operator=(const append_entries_request&) = delete;
        append_entries_request(append_entries_request&&) noexcept = default;
        append_entries_request&
        operator=(append_entries_request&&) noexcept = default;

        model::node_id node_id;
        protocol_metadata meta;
        model::record_batch_reader batches;
        flush_after_append flush; // <----------- New field
        static append_entries_request make_foreign(append_entries_request&& req) {
            return append_entries_request(
            req.node_id,
            std::move(req.meta),
            model::make_foreign_record_batch_reader(std::move(req.batches)),
            req.flush);
        }
    };
    ```

## Drawbacks

- When recovering followers for those partitions that use `quorum_ack` leader will
  have to wait before updating `commit_index` until it will eventually get
  information from the follower that entries were flushed.

- Adds additional 8 bytes per group to heartbeat response.

## Rationale and Alternatives

- There was a proposed alternative to use consistency level as a part of
  `append_entries_request` to let the follower know if it have to flush
  immediately. This would require persisting this information in log as entries
  send during recovery are read directly from the log/cache. Additionally it has
  no additional benefit as recovery can work in the same way for `quorum_ack`
  and relaxed consistency level

---

### Rfcs 20200529 Controller Refactoring (`docs/rfcs/20200529_controller_refactoring.md`)

- Feature Name: Controller Refactoring
- Status: in-progress
- Start Date: 2020-05-29
- Authors: Michal
- Issue: ch545

# Executive Summary

As redpanda feature set started to grow it became clear that our current
controller design became hard to maintain. In its current shape the controller has
lots of responsibilities and many dependencies. In order to move forward and
make it easier and faster to develop new feature we need controller refactoring
that will leverage recently available Raft state machine abstraction and
state snapshots. New controller design should be decoupled from other system
components and have clear set of dependencies. Additionally the current
implementation lacks sound concurrency control.

## What is being proposed

The proposition is to refactor/split the controller to meet the following
requirements:

- Sound concurrency control
- Ease of adding new features
- Decoupling and reduced number of dependencies
- Clear and reduced set of responsibilities
- Ability to handle long running tasks (f.e. topic resharding)

## Why (short reason)

Currently controller has the following responsibilities:

- joining cluster
- creating raft-0 group
- managing topics (create/delete operations)
- controlling partition assignment
- updating the partition allocator
- updating metadata cache with topic state changes
- updating metadata cache with cluster member changes
- forwarding leadership notifications from raft groups to metadata cache
- managing nodes connection cache

Additionally the controller current concurrency control has following drawbacks

- it is pessimistic i.e only one operation can happen at the time which may
  become a bottleneck in the future
- in order to be correct it has to block indefinitely (no timeouts are allowed)
  while waiting for entries to be replicated

Large set of responsibilities and complicated concurrency control make it hard 
to maintain and extend.

## How (short plan)

Proposed solution includes splitting controller into separate and mostly
decoupled modules that will have clear responsibility boundaries.
Looking at the current list of controller functions we can enumerate
following mostly separate modules:

- **cluster members management** - it includes adding/removing nodes from the
  cluster, providing information about members to Kafka APIs,managing internode
  connections, handling cluster join sequence

- **partitions leadership handling** - this includes handling leadership
  notifications, caching current partition leaders, triggering metadata
  dissemination

- **topics management** - the topics management includes keeping topics state,
  validating topic/partition requests, state reconciliation (for optimistic locking),
  providing topic related information for Kafka API

- **partitions placement constraint solving** - this includes the algorithm
  providing information about how to assign partitions to resources existing in
  redpanda cluster in optimal way

In proposed solution each module is responsible for its own state. Metadata
cache used currently by Kafka APIs is designed as a facade over all
above-mentioned modules.

## What use cases does it support?

Proposed solution is going to support all the use cases that current controller
implementation does. Moreover the new design will be easily extendible and able
to support long running jobs.

## What is the expected outcome?

After the refactoring we should end up with controller code that is clear and
easy to maintain.


# Detailed design - How it works

## `cluster::members`

Cluster members is a map containing all cluster brokers.
It provides information to Kafka Metadata API and other components about broker
existing in the cluster and all the details about the brokers
(ip addresses, number of cores, etc.).

## `cluster::members_manager`

Members manager class is responsible for updating information about cluster
members, joining the cluster and creating connections between node.
This class receives updates from controller STM. It reacts only on raft
configuration batch types. All the updates are propagated to core local
cluster::members instances. There is only one instance of members manager
running on core-0.

## `cluster::metadata_dissemination_service`

The metadata dissemination service is responsible for disseminating and
listening to information about partition leaders. It will also propagate those
information to the nodes as not all groups are instantiated on all of the nodes.
The metadata dissemination service using disjoint sets based dissemination
algorithm. The service sends updates to all core local instances of
`partition_leader_table`.

## `cluster::partition_leader_table`

Class holding mapping between NTP and partition leaders. This information can is
independent from the one contained in topic_table as leadership notification can
happen before the node knows about given topic. The `partition_leader_table`
expose an interface allowing users to wait for leaders to be elected.

## `cluster::topics_frontend`

The topics frontend is exposing an APU to manipulate topics and partitions.
The frontend in responsible for request validation and requesting partition
allocator to get the partition assignments. Topics frontend creates the command
and replicates them using `controller_stm`.

## `cluster::topic_table`

The topics state is a core local mirror copy of all topics and partition
properties. (sharding can be easily implemented in future). The topic state
accepts applies from `controller_stm` and on apply it calculates operation result.
As `topic_state` instance exists on every core and it contains the same data the
`controller_stm` have to apply updates to instance existing at every core.
In order to do that we use `core_mirrored_state` wrapper class. The wrapper
replicates commands application on every core.

### Note about ordering

`raft::state_machine` and `raft::mux_state_machine` executes only one apply at
the time (they wait for apply to finish). Thanks to that updates are applied in
the same order on every core as no other updates can interleave with then one
currently being applied to state.

## `cluster::controller_backend`

The `controller_backend` is a component responsible for actually applying changes
being reflected in the state to the cluster. Currently this involves topic
creation and deletion. Backend queries the state components for changes and when
they are available execute actions. The backend component has the housekeeping
timer that retries failed and not finished operations.

## `cluster::controller_stm`

The controller STM is going to be component handling all controller related
state replication and updates. Currently the only state that redpanda stores in
controller are topic configurations and partition assignments. This is going to
change in the future as we are probably going to extend redpanda with features
like ACLs authentication, etc. Those feature will require keeping cluster wide
state. Additionally controller_stm applies raft configuration batches to
`cluster::members_manager`.

## `cluster::partition_allocator`

Currently named `cluster::partition_allocator` it will be update with the
information coming from the cluster members manager. The partitions scheduler
will be responsible for solving the partitions placements constraints. i.e. it
will assign partitions to hardware resources in an optimal way taking given
constraints (f.e. number of cpus, disk space, free mem, etc.).

### Topics operations component diagram

```plain

+----------+                  +----------+                +----------------+
| External |      Request     |          |     Replicate  |                |
|   APIs   +----------------->+ Frontend +--------------->+ Controller STM |
|          |                  |          |                |                |
+----------+                  +----------+                +----------------+
                                                          |     Raft 0     |
                              +----------+                +-------+--------+
                              |          |         Apply          |
                              |  State   +<-----------------------+
                              |          |
                              +-----+----+
                                    ^
                                    |  Wait for changes
                                    |
            +------------+    +-----+----+    +------------+
            |  Partition |    |          |    |    Group   |
            |   Manager  +<---+ Backend  +--->+   manager  |
            |            |    |          |    |            |
            +------------+    +----------+    +------------+
```

### Concurrency control

When executing operations at frontend level we do not require locking. We relay 
on state updates to be correctly 

## Interaction with other features

### Metadata cache

In its current shape the metadata cache holds all the state that is required to
fill Kafka metadata response. The state is updated by the Controller and
metadata dissemination service. In order to simplify the design and to be in
align with the SRP (Single Responsibility Principle) in the proposed design the
MetadataCache will be the facade over other members of `cluster` namespace.
The metadata cache core-affinity will be independent from the actual state
location as the Metadata cache facade, for simplicity, is going to be
instantiated on every core. It will root requests to different core or use core
local whole state copy depending on the use case.

```plain

   Kafka API                  Kafka Proxy
       +                           +
       |                           |
       |                           |
       |                           |
       |                           |
 +-----v---------------------------v--------+
 |                                          |
 |         Metadata Cache (facade)          |
 |                                          |
 +----+-------------+----------------+------+
      |             |                |
 +----v----+   +----v-----+    +-----v------+
 | Members |   |  Topics  |    |  Leaders   |
 +---------+   +----------+    +------------+

```

### Kafka API

All components that are used by Kafka API requests are instantiated on each core.
Request dispatching to target core and cross core communication is transparent.
There are currently only two components that are going to be used by Kafka API
those are the `cluster::metadata_cache` and `cluster::topics_frontend`.

## Drawbacks

The main drawback of current solution is the number of component that have to
interact together in order to provide the controller features.

## Rationale and Alternatives

The main alternative to current design that was considered was using a
pessimistic locking approach. The pessimistic locking may at the first sight
seems easier to implement but considering its performance penalty and what is
more important the need to implement state reconciliation (similar to the one
that is required for optimistic locking)

## Unresolved questions

Should we distribute entities over caches or keep a copy per core as it was
proposed in the design.

## Telemetry & Observability

Each of the cluster state management component will expose set of domain specific
metrics. The metric set should allow to check rate of successful operations and
rate of errors.


## Additional resources

- [Cluster components diagram](https://docs.google.com/drawings/d/14UMFwXATAwSzDo9pZQX2VefMh3gsgX7r2BdguFDDH2U/edit?usp=sharing)
- [Topic creation sequence diagram](https://www.websequencediagrams.com/cgi-bin/cdraw?lz=CnBhcnRpY2lwYW50IGthZmthX2FwaQAJDXRvcGljc19mcm9udGVuZAAlDQA5BXRpb25fbGVhZGVycwANEV9hbGxvY2F0b3IAQRRzdGF0ZQB3DWJhY2sAVhBjb250cm9sbGVyLXN0bQCBJg1yYWZ0LTAKCm5vdGUgb3ZlcgAeDywAGwY6IFNUTQoAgVcJLT4AgUYPOiBjcmVhdGVfAIFoBSgpCgCBZg8tPgCBPg46IGFzc2lnbl8AgXgJcyhwbGFjZW1lbnRfY2ZnKQCCVAUAgXMKAF8TAIIyCgBGBm1lbnQAZhIAgW4OOiByZXBsaWNhdGVfYW5kX3dhaXQoAIEvDF9jbWQpCgCCIw4tPgCBfQgANAkAJg5vcCkKAIJDBgBUHHJlc3VsdDxvZmZzZXQ-ACkIAIEQEmFwcGx5KGJhdGNoAHgSAINsDAAjCAB7EQCEEQwAgmgSY29uZmlybQCEUQhpb24AIg8AgiMQZXJyYzo6c3VjY2VzcwB6GACEAAoAIQ4Ag3QVAIVmDTogd2FpdF9mb3IAhXcOAIYLDACEXhNuZXcAhiwJCgoKbG9vcCByZWNvbmNsaWxpAIFTBSBsb29wCiAgIACGDggAgjAQAHAJZGVsdGFzAB0OAIZDBzogY2hlY2sgaWYgAIZpBSBpcwCDHgVpZWQKZW5kAIVYEgCHfQkAhg0IAId8BiAAg3oGCg&s=default)

---

### Rfcs 20200713 Ghost Batches Recovery (`docs/rfcs/20200713_ghost_batches_recovery.md`)

- Feature Name: Compacted log recovery
- Status: draft
- Start Date: 2020-07-13
- Authors: michal
- Issue: 

# Executive Summary

In raft log maintained by the consensus algorithm is gap less. This assumption
allow nodes to reason about log correctness checking only the last log entry. This
assumption is valid for non compacted Redpanda log instances. Compaction
algorithm introduces intentional gaps in the log to remove outdated values.
Additionally the compaction is not coordinated on all nodes and can be in
different state at each replica.

Compacted log record batches (gaps) makes Raft recovery impossible, as follower
is unable to check if it log is the same as the leader one.
Moreover in Redpanda offsets are assigned to batches when they are appended
to disk. Having gaps in the log would lead to incorrect offset assignment as
even though there are gaps in the offsets follower will assign consecutive
offsets.

## Example

```plain
    Leader log:   [0,10][11,20]<--gap from 21 to 32-->[33,40]

    Follower log: [0,10]
```

When offsets are send to follower the offset assignment happen in its
log_appender instance. The offsets are assigned monotonically starting
from 0 without gaps.

During normal recovery procedure the follower would have the following offsets
assigned:

```plain
    [0,10][11,20][21,28]
```

this is incorrect as batch `[21,28]` has different offsets `[33,40]` at the 
leader.

### Legend

```plain

[base_offset,last_offset] - single batch

<-- --> - gap
```

## What is being proposed

The proposal is to use the 'ghost batches' indicating intentional gaps in the
log. The ghost batches will not be persisted to disk but will be used in raft
recovery mechanism to force correct offset assignment and provide gap-less log
for Raft follower nodes.

# Guide-level explanation

## What use cases does it support?

Proposed solution supports raft recovery for compacted logs not comprising the
Raft protocol consistency guarantees. The solution does not introduce additional
storage cost introducing only small overhead in append entries request
(57 bytes per one gap).

## What is the expected outcome?

Expected outcome of the proposed solution is to have mechanism allowing Raft to
work with compacted logs without comprising any of it safety guarantees.

## Introducing new named concepts

Proposed solution defines new `model::record_batch_type` being representation
of 'ghost' batch denoting an intentional gap left after compacting the log.

## Detailed design - What needs to change to get there

### `raft::recovery_stm`

In raft recovery state machine record batches read for the follower recovery are
materialized in memory as `ss::circular_buffer<model::record_batch>`.
The recovery STM iterates over materialized records buffer and inserts
`storage::ghost_record_batch_type` into places where gap in the log occur.

```c++
  // fill batches collection with gaps
  auto gap_filled_batches = details::fill_gaps(
              start_offset, std::move(batches));
  _base_batch_offset = gap_filled_batches.begin()->base_offset();
  _last_batch_offset = gap_filled_batches.back().last_offset();

  auto f_reader = model::make_foreign_record_batch_reader(
                      model::make_memory_record_batch_reader(
                                         std::move(gap_filled_batches)));
  return replicate(std::move(f_reader));
```

### `storage::disk_log_appender`

Disk log appender use ghost record batches for correct offsets accounting but
do not persists them to disk.

```c++
ss::future<ss::stop_iteration>
disk_log_appender::append_batch_to_segment(const model::record_batch& batch) {
    // ghost batch handling, it doesn't happen often so we can use unlikely
    if (unlikely(batch.header().type == storage::ghost_record_batch_type)) {
        _idx = batch.last_offset() + model::offset(1); // next base offset
        _last_offset = batch.last_offset();
        return ss::make_ready_future<ss::stop_iteration>(
          ss::stop_iteration::no);
    }
    return _seg->append(batch).then([this](append_result r) {
        _idx = r.last_offset + model::offset(1); // next base offset
        _byte_size += r.byte_size;
        // do not track base_offset, only the last one
        _last_offset = r.last_offset;
        auto& p = _log.get_probe();
        p.add_bytes_written(r.byte_size);
        p.batch_written();
        // substract the bytes from the append
        // take the min because _bytes_left_in_segment is optimistic
        _bytes_left_in_segment -= std::min(_bytes_left_in_segment, r.byte_size);
        return ss::stop_iteration::no;
    });
```

### Term handling

The ghost batches may fill gap that spans across batches coming from different 
terms.

Consider an example:

```plain
              |- term 1 --|                       |- term 3 ...
Leader log:   [0,10][11,20]<--gap from 21 to 32-->[33,40]
```

In this example batches in range `[0,20]` come from term `1` while batches in
range `[33,...]` from term 3. We are unable to determine what was the term of
that were removed during compaction process.
In order to guarantee correctness we assume that the gap has term of batch that
immediately follows it in the log. (In current example ghost batch in range
`[21,32]` would have term `3`). An alternative would be to use term of a batch
immediately preceding the gap in log, however the ghost batch may be the first
batch in the range of batches, then it term couldn't be established.
This problem do not exists for the proposed approach as the gap will never exist
as a last element in a series of batches delivered for recovery.

## Drawbacks

The main drawback of the solution is an overhead introduced by gap filling
batches i.e. 57 bytes per gap. In real life scenarios the overhead will be
minimized as some large gaps will require only single gap representing batch even
though many batches were compacted.

## Rationale and Alternatives

More complicated alternative would be to force follower assigned offsets with
those coming from the leader. The main flaw of this solution is the fact that it
is hard to correctly reason about it's impact on raft protocol safety as it
diverge from the originally published Raft algorithm.

## Unresolved questions

---

### Rfcs 20201208 Fragmentation (`docs/rfcs/20201208_fragmentation.md`)

- Feature Name: Fragmentation Analysis
- Status: draft
- Start Date: 2020-12-08
- Authors: Travis Downs
- Issue: https://github.com/vectorizedio/Redpanda/issues/211

# Executive Summary

Fragmentation of the Seastar allocator, caused by a combination of factors such as inherent susceptibility of non-moving allocators to fragmentation as well as the use of opportunistic caches that can fill the available memory, can result in even modestly large allocations failing (as low as a "few MB"), and generally in a failure of the shard and process.

This RFC does not propose one specific new feature that will _solve_ fragmentation, per se. A non-moving allocator will always be subject to fragmentation. Rather, the problem is outlined and a variety of mitigations and partial solutions are presented for analysis.

The scope of this RFC is the main Redpanda binary written in C++ and using the Seastar thread-per-core event-driven framework and allocators.

## What is being proposed

The fragmentation problem is described, analyzed and finally possible mitigation or solutions are discussed.

## Why (short reason)

Fragmentation resulting in failed allocations is a serious functional (crash) and performance risk to the Redpanda cluster.

## How (short plan)

We don't yet know how, it needs to be discussed.

## Impact

Solution dependent. Specific impacts are discussed for each approach below.

# Motivation

## Why are we doing this?

Fragmentation must be understood and where possible addressed, as the symptom may be a failure of a shard, node, or the entire cluster, depending on the nature of the load impacted by fragmentation

The nature of fragmentation makes its occurrence is difficult to predict and when it occurs there may be few workarounds short of a code change. The impact can range from cascading node failures, to serious performance impacts (due to the interaction with reclaim). Furthermore, this may present a DOS (denial of service) attack vector for any party who can access a Redpanda cluster, since it is entirely plausible that the pattern of allocations can be controlled by the attacked (such control would very hard to avoid).

Fragmentation is a pervasive concern. There are many thousands of existing allocation points in the code, and these will scale proportionally as more code is added. A developer cannot be expected to think of of the consequences of fragmentation each time _potentially allocating_ code is added: there are just too many and fragmentation is too complicated. Contrast this with e.g., adding locking via a mutex: these are few enough and the rules are clear enough that we could expect good "mutex discipline". Allocation is different. Finally, many allocation sites are invisible to the developer, hidden by standard library or third party components.

Redpanda is position as a high-performance, high-availability service and fragmentation puts this at risk.

## What is the expected outcome?

A comprehensive (?) plan to manage and mitigate fragmentation. At the very least, a better understanding of fragmentation.

# Reference-level explanation

## Named Concepts

We define some terms that may not be familiar to all readers, or which I have possibly just made up on the spot, here.

**Span:** In the Seastar allocator, and allocators more generally, a _span_ is a contiguous range of 1 or more pages, treated as a unit for some operations. In the specific case of Seastar the buddy allocator uses spans of size 2^n. The `Seastar::page` struct serves as the type for both pages and spans: the first page in a span (and to a lesser extent the last) holds information for the whole span.

**Non-moving allocator:** A non-moving allocator is one which cannot move (in the virtual address space) the pointers it returns. Typical allocators in native languages, like `malloc` in C/C++ and `new` in C++ are
non-moving: they return a pointer value and the memory must remain allocated at that location for the lifetime of the pointer. This is opposed to moving (or _compacting_) allocators, which may move memory they have returned. Garbage collected languages often feature moving allocators, and such allocators are possible in native languages, e.g., using _handles_ (double indirection) rather than pointers.

**Availability %:** This is a measure of fragmentation and applies to a specific potential allocation (span) size. It is calculated as: the fraction of _free memory_ which can be used to satisfy allocations of the given size (not considering reclaim). For example, if there is 100 MiB of free memory, but only three 1 MiB blocks can be allocated before failure (or reclaim), the availability is 3% for 1 MiB spans. An availability of 0% indicates complete fragmentation at a given size.

**High Water Mark:** An indication of the highest part of the heap that has been _touched_ by allocation (assuming spans are allocated from lowest addresses first, which they are in the Seastar allocator). The high water mark is somewhat larger than simply the largest live set of allocated memory at any point in the past because of fragmentation (not all pages below the high water mark are actually filled).

**Reclaim:** The process by which the Seastar allocator requests the application to free memory, which may be asynchronous or synchronous.

## Interaction With Other Features

Fragmentation as discussed is a cross-cutting concern which interacts with almost any feature that allocates memory, particularly anything which might request large contiguous allocations, such as a large `std::vector`. The concern is pervasive because it is difficult inspect the implementation of every class to understand its allocation behavior, especially the significant amount of third party code we use, and it is also difficult to determine what the maximum expected size of any particular contiguous allocation will be, even if the allocation points are identified.

## Telemetry & Observability

Some of the solutions relate specifically to observability, and they are discussed in that section.

## Seastar Allocator Background

This section has a comprehensive(ish) description of the Seastar allocator. This is necessary because the design of the Seastar allocator is tied to behavior of fragmentation.

### Overall Approach

In its recommended configuration, Seastar uses its own custom allocator implementation, completely overriding the standard allocation routines, `malloc`, `free`, `operator new`, `operator delete` and so on. Functionally, this is generally transparent to applications built on Seastar (although good performance requires following some rules), and in fact the Seastar allocator could be disabled by defining the `Seastar_DEFAULT_ALLOCATOR` macro at compile time to use the usual allocators (we do this in debug configuration, to enable ASAN).

Al memory handling is implemented in in `memory.cc` and associated header `memory.h`. All memory for each reactor thread is tracked in a `thread_local` `cpu_pages` object.

All memory allocations are carved out of a large virtual address space allocation of size 2^44 (~17.6 terabytes), reserved at startup: each core gets an equally sized partition of 2^36 bytes (~68 GB) within that space (see `cpu_pages::initialize()` and `mem_base()`). Yes, these values are currently hard-coded, so no reactor thread can easily use more than 68 GB of memory. 2 MB hugepages are requested for the entire region with `madvise`, and the allocator also supports hugetlbfs. When the allocator is configured (see `ss::memory::configure()`), a specific amount of memory is `mmap`ed (but not `MAP_POPULATE`d) within the virtual allocation. The per-page state for each page in this allocation is tracked in the cpu_pages::pages array, which is embedded directly at the start of the memory for the core (so some number of initial pages are reserved for this purpose and never available for allocation).

Every allocator is core-local, implemented by making the core allocator structures described above (and below) `thread_local`. This means that allocations must be (internally) freed on the core on which they were allocated, and the large majority of allocations are expected to live only on the owning core. User code calling `free` on the "wrong" thread _is_ supported but is inefficient: requiring that the pointer be sent to the correct core using message passing (see `cpu_pages::free_cross_cpu` and associated methods).

The base unit of allocation is the _page_, 4096 bytes by default (can be overridden at compile time). This happens to be the same size as the smallest page size on x86, but they are not intrinsically related. For example, it would be [possible](https://github.com/scylladb/Seastar/issues/836#issuecomment-748531815) to increase the Seastar page size to 16K regardless of the OS page size.

The main entry point for all allocations is `ss::memory::allocate(size_t)`. All allocations are either _small_ or _large_, with the dividing point being hard-coded at 4 pages (16 KiB). Small allocations go to a dedicated small allocator which is _pool-based_ and is backed by the large allocator, while large allocations are serviced by the buddy allocator directly (hereafter we might use the terms _large allocator_ and _buddy allocator_ interchangeably).

We cover the large pool first, then the small pool (since some large pool concepts are useful to understand the small pool).

### Large Allocations

Large allocations use a buddy allocator system: the entire committed region of memory starts out as a single large span, and each 2^n-sized region of memory can be split into two 2^n-1 sized regions during allocation to satisfy a smaller allocation if no smaller spans are available, or two adjacent regions ("aligned", i.e., buddies) can merged into one 2^n+1 region. Those basic operations, which can be applied recursively, pretty much define the entire behavior of the buddy allocator.

Large allocations are rounded up to the next multiple of the page size (4K), and eventually end up in `cpu_pages::allocate_large_and_trim`. This function uses the free_spans list, starting with the smallest suitable span, to find the smallest free span that can fit the requested allocation. If the span chosen wasn't the smallest possible (i.e., wasn't the requested size rounded up to the next power of 2), the remaining parts of the span are split off and returned the appropriate free list. For example, if a request for a 7K allocation is satisfied by a 32K span, an 8K span is actually returned to the caller and one 8K and one 16K span comprising the salvageable unused memory are returned to the buddy allocator.

Freeing eagerly merges the returned allocation with its buddy if both are free, recursively. This means that the large heap is always in a most-merged state: the number of spans is as few as possible and they are as large as possible.

Available (free) spans are kept in linked lists (`cpu_pages::free_spans`): one for each size. When a request for a span of a given size occurs, the first free span in the list is returned, if any. Otherwise, larger lists are searched, starting with the next larger size and any found larger span will be split. Freeing adds the span to the front of the free span list (or at least it adds _some_ span, after merging). This whole paragraph is just trying to convey the idea that spans are added and removed from the free lists in _FIFO_ order: there is no particular effort expended to defragment the freelist. This means that in some cases where fragmentation could be avoided by a best-case allocator it will be not be avoided in Seastar because the freelist behavior is not fragmentation aware.

Like most userspace allocators in native language, the buddy allocator is a _non-moving_ allocator.

#### Fragmentation in the Large Allocator

Fragmentation is a problem in the large allocator. As shown in the experiments below, random allocation of smaller blocks can prevent larger allocations completely. Specifically, after the memory is filled with objects of a given size, even after ~97% of the memory is freed, allocations 7 to 8 orders larger will often not be available.

We give more details on _how_ fragmentation occurs below.

### Small Allocations

Small allocations (see `cpu_pages::allocate_small()`) are mapped onto a size class using a log-linear mapping: there are size classes for every power-of-two size, as well as three intermediate sizes between each power of two. E.g., there are 256-byte and 512-byte size classes, as well as for three intermediate sizes 320, 384 and 448 linearly spaced between those two points. Each size class maps to a pool allocator that only allocates objects of that size.

Each small pool has a _preferred_ and _fallback_ span size. The fallback size is simply the smallest span size that is equal or larger to a single object in the pool. Using this size might have a lot of waste: imagine a pool for 5120 byte objects, which would have a minimum (fallback) size of 8 KiB but this can fit only a single object and would waste 37.5% of the span. Instead a 32 KiB span is chosen as the preferred size, with a waste of only 6.25%.

Each small pool is organized as a free list of single objects as a linked list, with a target count _min_size and _max_size. Allocation requests are allocated from and freed to the free list, in an LRU fashion (i.e., the most recently freed object will be used to satisfy the next allocation). Memory for individual small pools is obtained from the large allocation path, in chunks of `_span_sizes.preferred` pages (or `_span_sizes.fallback` if that fails). This means that the smallest spans allocated by the large path are actually those needed to satisfy small pool refills, since these are as small as 1 page, whereas large-path allocations to satisfy requests directly will be at least 5 pages (since <= 4 pages are handled by the small path).

This behavior produces a significant link between small memory allocations and requested spans from the large allocator. For example, objects of 1024 (or slightly less) bytes use a preferred span of 4K, so will allocate spans of 4K. Objects of 1025 bytes (up to 1280 bytes) have a preferred span of 20K, so will allocate in a different pattern and affect the large allocator differently. Generally speaking, it seems that the small pools could reduce fragmentation by preferring larger spans of a more consistent power-of-two size.

Each small pool tries to keep the size of the free list between the `_min_free` and `_max_free` sizes: when it exceeds the max size, `trim_free_list` is called. This pops objects off the primary pool free list and adds them to a per-span freelist (`page::freelist`). If ever a span is completely freed (i.e., all the objects associated with the span end up on the per-span freelist), the entire span is returned to the buddy allocator. This is the only way memory ever escapes a given small pool.

#### Fragmentation in the Small Pool

The small pool drives allocations in the large pool, so it causes fragmentation in the same way as allocations directly to the large pool, with varying sizes as set by the preferred size for each pool.

In addition, the small pool is subject to its own kind of fragmentation, where spans used by a given small pool can never be used by allocations for objects of other sizes, if even a single object remains allocated in the span. This is different from the buddy allocator behavior, which can suffer fragmentation but where all unallocated memory is still available to other allocations of any size (as long as the span is large enough).

This could be an issue if many allocations of a specific size happen in some phase of operation, where after most but not all are freed, and then this size becomes unpopular. A large amount of memory could be used by the unpopular object size (and, of course, this process can repeat with other pool sizes). I don't consider this as serious of a problem as general buddy allocator fragmentation since (a) this same problem exists in general purpose allocators and usually doesn't cause exhaustion, probably because the allocation pattern that would cause a serious amount of waste is unusual and (b) the allocations made by the things that tend to fill up memory are large enough that they are allocated directly in the large pool.

### Reclaim

A specific feature of the Seastar allocator is that when low on memory it can make an _upcall_ to the application code to request that the application free up memory -- a so-called _reclaim_ request (see `ss::memory::reclaimer`).

Originally, this only supported an asynchronous mode, where the upcall would inform the application of the low memory condition, which would respond with a callback to be scheduled later which would try to free memory. This didn't always work well (memory might run out before the reclaim got scheduled, or the reclaim wouldn't free enough, perhaps because it didn't know how much was needed).

To help solve some of those problems a _synchronous_ mode was added: the upcall is made to get a reclaimer which is immediately invoked to attempt to free memory before the allocation attempt proceeds. Redpanda only uses synchronous reclamation, and so we only discuss this type below.

Seastar runs reclaim at two different times, described below.

First in `cpu_pages::maybe_reclaim` (called every time an allocation occurs) if the current free memory is less than `min_free_pages` (defaults to 20 MiB and Redpanda doesn't change this default) then synchronous reclaim is run once (and the result is ignored). It is important to note that the 20 MiB threshold here is just total free pages: it doesn't mean that the _high water mark_ will stay 20 MiB below the full region, because fragmentation means the entire region may be filled before the threshold is breached. In fact, memory allocation may fail (triggering the second reclaim path) before this reclaim path is ever used.

Second, when an allocation actually fails (i.e., there are no spans of sufficient size), reclaim is attempted. Unlike the first case, reclaim will occur repeatedly (see `cpu_pages::find_and_unlink_span_reclaiming`) while progress being made (as determined by the reclaim result of `reclaimed_something`) until the allocation succeeds. The allocator tells the reclaimer how much memory it needs, but even if the reclaimer satisfies the request, allocation may keep failing due to fragmentation. Specifically, the allocator needs N contiguous pages, and asks the reclaimer for N pages, but these N reclaimed pages may not be contiguous (and in general there is no reason to expect them to be). This means that before an allocation ultimately fails, it will first flush all memory subject to reclaim.

In Redpanda, there is one reclaimer: the batch cache (see `batch_cache::reclaim`). This iterates through the records in least recently used order, applying some rules about what can be reclaimed until it reclaims enough bytes. It also has rules to reclaim _more_ than the requested number of bytes, if the requested number is below some threshold, or there have been frequent reclaim requests recently.

## Experiments

### Synthetic Fragmentation Tests

I did some synthetic fragmentation tests to observe fragmentation behavior. Most of the results were mostly as expected from theory, although some interesting things did crop up due to the specific behavior of the Seastar allocator.

#### Worst Case Fragmentation

The worst case fragmentation is easy to calculate and is not specific to the buddy allocator, but applies to any _non-moving_ allocator. Fragmentation isn't a scalar value, but a spectrum that depends on the desired allocation size `S`. The worst case occurs when all free memory occurs in spans of slightly less than `S`. It is always possible to set up such a scenario: allocate all available memory with the smallest possible blocks of size `M`, then in every contiguous region of size `S` free all but one block.

This results in free to total memory ratio of `(S - M) / S` but no way to allocate any `S`-sized blocks. As an example, a block of 1 MiB may fail to allocate even when memory is 96.8% free if the used memory is in carefully spaced 32 KiB blocks.

Tests `pattern_alloc_small` and `pattern_alloc_big` illustrate this in practice, by allocating blocks in the pattern described above. The small tests allocates blocks that use the small pools and demonstrates that this problem is somehow _worse_ in the small pools: memory can be fragmented within the spans allocated to a given small pool, such that the small pool consumes a large number of pages for a small amount of actually allocated memory, and this space is only available to allocations of exactly that size. This is unlike the buddy allocator which suffers from fragmentation but at least allows any allocation that can fit in any free span to succeed.

We aren't likely to encounter _worst_ case fragmentation outside of malicious input (which is a possibility not be dismissed), but it serves as a useful upper bound on fragmentation, and is easy to demonstrate.

#### Randomized Fragmentation

As a somewhat more realistic synthetic test we can allocate most of the shard's available memory (as the batch cache tends to do) and then free a specified fraction of it at random. This is probably not actually a _terrible_ approximation to what the buddy heap will look like after a lot of uptime, as the mechanics of the buddy allocator mean that the general trend is towards more entropy in the heap.

If we fill the available memory with 32 KiB allocations, then free 96.8% of them randomly (this fraction chosen to line up with the worse case example above) we find the following availability of various chunk sizes:

~~~
Used memory:    151 M
Free memory:   3432 M
Page spans:
index   size    total   used    avail%  spans
0       4K      404K    400K    100.0   101
1       8K      280K    280K    100.0   35
2       16K     944K    928K    100.0   59
3       32K     225M    115M    100.0   7k
4       64K     212M    832K     96.8   3k
5       128K    379M    384K     90.6   3k
6       256K    619M    0B       79.6   2k
7       512K    859M    0B       61.5   2k
8       1M      770M    0B       36.5   770
9       2M      414M    2M       14.1   207
10      4M      72M     0B        2.1   18
11      8M      0B      0B        0.0   0
12      16M     0B      0B        0.0   0
13      32M     32M     32M       0.0   1
14      64M     0B      0B        0.0   0
15      128M    0B      0B        0.0   0
16      256M    0B      0B        0.0   0
[... all remaining spans empty]
~~~

In this case, unlike the worst case, we _are_ able to allocate some 1M spans: fully 36% of the free memory is available for 1M allocations. From there, however, it gets worse quickly: only 14% and 2.1% of the free memory is available for 2M and 4M spans. For 8M spans, we are out of luck: nothing is available. Yes, we have 3.4G of free memory, but can't allocate a measly 8M.

So the random case isn't too far off the worst case: it only bought us 3 additional orders of allocation. This pattern repeats itself with other parameters.

The results above follow directly from the math of random allocations. The chance that a span of length `S` is fully free when considered as made of blocks of size `M`, given that the fraction of free memory is `F`, is `F^(S/M)`. This closely matches the availability results above (the divergence is probably related to the fact that some blocks other than those allocated by the simulation are also present). In order to calculate the _expected_ number of blocks of a given size (rather than simply the availability %) we need to multiply by the total number of possible blocks (where `T` is the total memory): `F^(S/M)*T/S`. This calculation shows that the expected number of 8M blocks is about 0.1 in this scenario: so we weren't that _close_ to having an 8M block available.

## Why (Not) Fragmentation

The examples above illustrate that fragmentation can be severe and occur with only a single allocation pass followed by a de-allocation pass.

Actual fragmentation will not be so severe, at least initially. For example, allocating N blocks of N size, then freeing those N blocks will return to the original state: fragmentation did not increase. Furthermore, interleaved allocations and freeing of blocks of various sizes, e.g., in a loop, will tend to only use a small working set of spans (equal to the highest number of simultaneously live allocations for each size class) as the allocators are FIFO.

For serious (read: allocation failures) fragmentation to occur, at least the following elements need to be present:

 1. Substantially all of involved memory region needs to be filled. Not necessarily "densely" - but the high water mark of allocation needs to reach close to the highest address. That is, if you imagine spans being used starting from the lowest addresses, the high water mark is the highest address allocated. If a shard has a small working set, the high water mark will stay small, because of the FIFO nature of the buddy allocator. If the entire per-shard region hasn't been filled, allocations can't fail since the untouched spans at the top of the region will be available.
 2. There needs to be a significant difference in size of the most commonly allocated objects and the allocation that fails. As a degenerate case we can see that if all allocations of the same size, fragmentation isn't possible: every available span will be of the required size, or larger. Similarly, for spans twice as large as the common size, you only need to find _two_ adjacent spans of the smaller size that are free for allocation to succeed, which is highly probable. In general, allocations that are are within "a few" orders of the dominant allocation sizes will succeed. In real applications there won't necessarily be a specific dominant size but a spectrum of sizes, but the same principle applies.
 3. Allocations of different lifetimes need to be interleaved. Without interleaving, groups of smaller allocations will tend to act like one large block, preventing fragmentation. A large number of small allocations may occur, but if they are freed around the same time the holes that cause a problem won't exist. It only takes a small amount of interleaving, however, to eventually fragment memory, because once some interleaving occurs, it tends to persist as a block with a different lifetime from the surrounding blocks will be freed at a different time, end up on the free list and reallocated a different time (potentially as part of a stream of allocations that would otherwise be contiguous).
 4. The total size of the available region must be fixed. If not, fragmentation can still happen in a sense (some free regions are unusable for some allocations), but it doesn't necessarily cause allocation failure as we can simply grow the region and allocate in fresh space. This is why most 64-bit userspace processes don't suffer from fragmentation: they can simply increase their virtual memory use, without increasing their physical memory use, by remapping pages from unusable holes to new contiguous VA space as described below.
 5. The allocator is not able to compact allocated memory. For example allocators in garbage collected languages can often move memory, which allows them to avoid fragmentation simply by copying live allocations to a contiguous region, removing any holes.

This list handily explains why fragmentation isn't something that affects every application. First, a large class of non-native applications written in garbage collected languages sidestep the whole problem by moving allocated memory during collection.

Then, for native applications, fragmentation _can_ occur, but new virtual address space can simply be allocated by `mmap` (or `sbrk`). This is actually a combination of items (4) and (5) above: the total size of virtual address space is not fixed, but you'd still eventually run out of physical memory if you left the fragmented pages allocated: but the magic of physical to virtual means that physical pages can be remapped to the new virtual pages. So even though pointers can't move in the virtual address space, the underlying physical pages can move (although this is obviously restricted to page granularity).

Many native applications also don't satisfy (1): they don't use all memory available to them (which, by default, is "all" memory). The memory used might be proportional to the input, or concurrent requests, or whatever other factors. If all of memory is every used, the application might be more likely to crash due to memory exhaustion (which _is_ common in native applications) than fragmentation, since if you've used 99% of memory, the next step is probably using 101%.

### Redpanda

In the case of Redpanda, however, all of these criteria are met. Most of them flow from the design of the Seastar memory allocator: it uses a relatively small (few GB) _fixed in the virtual address space_ per-shard region (4) which is not able to move memory (5). Allocation of different sizes (3) occurs frequently for many reasons, including simply the natural behavior of the code and also because of the rapid scheduling of coroutines performing different tasks on the same shard. Filling all of memory (1) occurs naturally in Redpanda because we let the batch cache grow to fill member (why not, it would be wasted otherwise) and rely on the reclaim mechanism to free up memory as needed.

## Mitigation Design

Here we examine several possible solutions to the fragmentation problem.

We broadly group them into avoidance, allocator modifications and application (redpanda) modifications.

### Avoid

Broadly speaking, the solution here is: *don't try to avoid fragmentation, but rather try to avoid failure even when fragmentation occurs*. The main approach is to simply to _avoid large allocations_.

This is the same approach taken by the Linux kernel (which uses a buddy allocator) for `kmalloc` (the guideline is expect allocations greater than order 4 to fail and for a long time the maximum allocation size was 128 KiB) as well as ScyllaDB (see for example issues [#7364](https://github.com/scylladb/scylla/issues/7364), [#6376](https://github.com/scylladb/scylla/issues/6376), [#108](https://github.com/scylladb/scylla/issues/108), [#1546](https://github.com/scylladb/scylla/issues/1546) and many more).

The main tools to _detect_ such cases are using the large allocation warning emitted by Seastar (not currently configured in Redpanda), and the diagnostics when an allocation fails.

Once a large allocation is detected it, avoiding it depends on the context (browsing the related ScyllaDB issues shows the large variety of approaches taken). Sometimes it is as simple as using a fragmented data structure (linked list, iobuf, deque) rather than a contiguous one. In other cases it was not expected that such a structure became so large in the first place, so some other adjustment is needed.

#### Drawbacks

A significant downside of the this approach is that is mostly reactive: problems are identified only when they cause a crash or (hopefully) when the warning is detected and acted on. Although the issue may be resolved, significant downtime may already have been incurred.

Another issue is that many allocations occur in third party libraries rather than the core code: although we have the source to all third parties and can patch them, this creates a divergence with upstream, creates a maintenance burden, and so on. Furthermore, it may be harder to avoid large allocations in third party libraries since we won't the same tools available to use as

#### Recommendations

At least, we should enable the large allocation warning, perhaps at 256 KiB. We can have it enabled in testing with a post-processing phase which determines if the discovered large allocations are from new locations which need to vetted to check how large they can get.

We should expose metrics related to fragmentation such as availability % for certain sizes in our metrics and possibly in telemetry, in order to understand what "typical" fragmentation looks like.

### Make Reclaim Span-Aware

As described above, the current reclaim mechanism means that reclaimers aren't aware of the required *size* for the pending allocation. So they may repeatedly free scattered chunks of memory which can't satisfy the current allocation. In the worst case, we can free almost the entire batch cache until we finally get a large enough contiguous chunk of memory. This means that fragmentation can introduce a tricky "semi-failure" mode rather than total failure (failed allocations). In this mode, performance is badly degraded because the batch cache is continually forced to a small size in order to satisfy large allocations inefficiently.

To mitigate this, we could implement a span-aware reclaimer: the reclaim code already tells the reclaimer the size of the allocation, but now the reclaimer should try to free spans that will result in a _contiguous_ region of at least the given size (rather than just a free of scattered regions reaching this total). For example, rather than LRU eviction, use an eviction algorithm which evicts entries which happen to be contiguous, or perhaps a hybrid algorithm which considers both the contiguity and age of the entries. This could significantly reduce the impact of fragmentation since the batch cache would suffer fewer evictions while reclaiming.

One can imagine two variants of this approach: in the simplest version, the reclaimer isn't aware of which spans are free from the PoV of the allocator and tries to return a contiguous region solely by freeing memory it owns. The potential problem is that with limited visibility opportunities to combine a newly freed region and an already free region will be missed, because the reclaimer doesn't know which regions are free. A more complete solution would combine free-region information exported by the allocator with reclaimer information to find contiguous regions.

### Allow the Batch Cache to Compact Memory

In a similar vein to the mitigation discussed above, we could make the batch cache capable of *compacting* the memory it uses. By tracking the pointers into the batch cache, we could move batches which are not currently being referenced to make them contiguous. This would greatly reduce fragmentation in the batch cache, the largest user of long-lived memory. This would require having a mechanism to understand when a batch is being referenced by a in-progress request, since we cannot safely move those batches. Perhaps this is similar to shared pointer tracking.

### Modify the Allocator

Another strategy would be to modify the Seastar allocator to improve fragmentation.

Per [this discussion](https://github.com/scylladb/seastar/issues/836#issuecomment-742924680) one small thing we can do reduce worst-case fragmentation is to increase the effective minimum span size to 16K. Currently, the large allocator only handles requests of at least that size already, but for some pool sizes the small allocator can request spans as small as 4K (but for many sizes it requests larger spans because of the _preferred size_ mechanism described above). Upping the minimum size of small page span requests would waste some memory in the small pool (due to partially filled spans), but overall this should _save_ memory since it would reduce the size of the pages array by ~75% per [Avi's comment](https://github.com/scylladb/seastar/issues/836#issuecomment-748531815): this would save perhaps 7M, much larger than the worst-case small pool waste of less than 1M.

This won't "solve" fragmentation since there are still pages as small as 16K, but it is easy and seems to have no downside.

More radical changes are possible. For example the allocator could reserve some fraction of the memory for each shard as a "huge pool". This wouldn't be used by the buddy allocator, but rather to satisfy allocations that might otherwise fail due to fragmentation. This allocator would be a more "traditional" `mmap`-based allocator which isn't restricted to a 1:1 relationship between VA and PA, and so doesn't suffer as much from physical memory fragmentation (since all allocations are multiples of 4K and 4K pages can be remapped freely by the OS). There are many variations here:

 - Is the huge allocator used unconditionally for allocations above some size, or only when the buddy allocator fails?
 - Is the huge pool per-shard or shared?
 - Do we just use the native built-in libc allocator (via the saved malloc pointer) as the huge poor, or layer another allocator on top of the reserved storage?
 - How exactly does the allocator reserve physical memory, since they mostly work in the virtual address domain? We need an allocator which carefully limits its physical memory use, but tries to use all of it e.g., by punching holes (e.g., with `MADV_DONTNEED`) in existing mappings.


### Abandon the Allocator

We know seastar works with the default system allocator: that's how we run it in debug mode (so we can use ASAN and friends). So in principle, we could use the system allocator, or some enhanced drop-in replacement instead. However, we would lose the significant benefits of the seastar allocator:

 - The performance benefits of per-shard allocation such as no locks or thread-local caches, exact NUMA affinity, etc.
 - Allocation may block the reactor threads: the seastar allocator minimizes the chance of blocking my allocating everything up front, but default allocators calling mmap may block in some cases.
 - The reclaim mechanism.
 - Heap profiling (though the replacement may offer similar functionality).
 - The exact division of memory among shards and the exact up-front nature of memory allocation. General purpose allocators tend to keep requesting more memory from the OS until something breaks, while the seastar allocator fails immediately if the assigned space is exhausted (after reclaim).

This would be a significant and potentially risky change.

### Quality of Life

Even though we might be living with fragmentation, we can make some quality of life improvements to make our life easier when root-causing failures.

 - OOM error messages should clearly log relevant details such as the "availability" metric discussed above to pinpoint if the failure was caused by fragmentation.
 - OOM error messages should include details of reclaim activity (if any) done during the failing allocation.
 - We should surface fragmentation related metrics (perhaps availability at a few key sizes) in our metrics.
 - We should surface metrics related to reclaim and batch cache activity to detect the "partial failure" case where the batch cache is thrashing described above (needs some thought about heuristics to detect this case).
 - We should track the high water mark in the seastar allocator and expose it, so we can see how it evolves over time.

## Unresolved questions

 - Is the underlying cause of fragmentation scattered across many allocations on many different paths, or are there a few "hotspots" which behave differently and tend to cause most of the fragmentation.
 - How does fragmentation evolve over time with typical loads? Do we quickly reach a plateau or does it continue to worsen for a long period of time?
 - In normal, high-load operation, are we continually "dodging a bullet", i.e., there is some small but non-zero chance of total failure due to fragmentation, or is it really only the specific workloads where we've recorded failures that can fail in practice? Said another way, should we expect a small stream of fragmentation related failures on high-load systems even if they don't do the identified problematic workloads?

---

### Rfcs 20210223 External Connectivity In K8s (`docs/rfcs/20210223_external_connectivity_in_k8s.md`)

- Feature Name: External connectivity in Kubernetes environment
- Status: in-progress
- Start Date: 2021-02-23
- Authors: RafalKorepta
- Issue: none

# Executive Summary

To give users out of the box experience Redpanda
Kubernetes operator should pave the way for the networking stack to
expose each Redpanda node in k8s environment.

## What is being proposed

The most performant and cost-efficient solution is to expose each
process to the outside, by specifying
[host port in the container ports specification](https://Kubernetes.io/docs/reference/generated/Kubernetes-api/v1.20/#containerport-v1-core)
for each Redpanda node.

## Why (short reason)

The Redpanda is focused on performance and its low tail latency. To
not add more network hops this proposal tries to merit dynamic
Kubernetes environment with the direct access to Redpanda process.

## How (short plan)

Service of type NodePort will have assigned port that is free
on each node and will be used in the advertised kafka API
configuration. The node port will be set in the host port
configuration of Redpanda container along with internal, RPC and
Admin ports. The external Kafka API port will be not configurable
and arbitrarily chosen to internal Kafka API + 1. Init container will
discover public IP of the host on which it was scheduled. The port
will be set in the advertised Kafka API address.

## Impact

The Kubernetes worker nodes must open port per Redpanda cluster
which might be an additional attack vector. The additional
firewalls should be managed to allow node access.

# Motivation
## Why are we doing this?

Redpanda can be easily managed in multiple environments, but vast
range of users plan to manage their workload in Kubernetes
environment. Saying that, teams would like to not introduce
clients application into their Kubernetes clusters. The Kafka API
requires that each broker will be reachable at any given time.
Canonical way of exposing webservers in Kubernetes environment is
to create Ingress resources that can be reconciled into cloud
provider load balancer. Another way of exposing different kind of
workloads is to expose their TCP port using Service type LoadBalancer
and again cloud provider load balancer will be provisioned. On
the other hand the TCP port can be exposed on each Kubernetes worker
node using Service type Node Port. All the mentioned methods
doesn't guarantee that each Redpanda node will be accessible
individually. One could create Service type Node Port for each
broker, but that would be a waste of ports in the whole cluster.
The optimal solution would be to pick one port and take public
IP of a node on which Redpanda is scheduled and use it in
advertised Kafka API.

During testing in AWS environment the service type Node Port doesn't
have assigned external IP. That's why init container needs to
retrieve node information from the Kubernetes API and use it in
advertised Kafka API address.

The operator should accommodate dynamic nature of any cloud
where nodes can disappear or be rescheduled.

## What use cases does it support?

Exposing each Redpanda node to networks outside the Kubernetes
cluster.

## What is the expected outcome?

The Redpanda operator will make sure that from the network standpoint
brokers will be reachable. Users need to manage security of the
exposed Kubernetes worker nodes e.g. `aws security groups`.

# Guide-level explanation
## How do we teach this?

The Kafka API requires that each broker will be reachable from
the client side. To expose Redpanda nodes, but not add additional
latency to the communication each Redpanda process will have assigned
unique host port. In case of cloud providers the host VM need to be
publicly reachable and have opened some range ports. The operator
would be responsible to orchestrate Kubernetes to achieve that.

## Dictionary

- Kubernetes Service type NodePort with empty selector that gives
  unused port that is available across all k8s worker nodes
- ServiceAccount with RBAC rule assigned to init container will
  allow retrieving the node information from Kubernetes API server

# Reference-level explanation

( This is the technical portion of the RFC )

## Interaction with other features

- The kafka multi listener needs to be set up in each Redpanda
  node, so that internal connectivity can stay as is. For external
  connectivity the dedicated port would be opened.
- The statefulset must bootstrap Redpanda cluster with or without
  external connectivity.
- The user intention defined in cluster.redpanda.vectorized.io custom
  resource should be reconciled into set of ready to use address with
  ports that can be reach from the internet. The status field will
  be the place to look for the addresses.

## Telemetry & Observability

- metrics that can be anchor in Redpanda or in VM network stack
    - Number of active connection
    - Number of dropped connection
    - Number of bytes processed
- Tail latency observed by the clients
- Additional probe that will live close to operator, but will try
  reach the Redpanda Nodes by their external address

## Corner cases dissected by example.

None.

## Detailed design - What needs to change to get there

The host port need to be added in kafka port in the Statefulset
definition. Service type node port needs to be added. The init
container needs to know how to get the public IP of the node.
New ServiceAccount need to be added with the RBAC rule for getting
node information.

## Detailed design - How it works

Each Kubernetes node needs to be exposed to the internet in order
to expose Redpanda node. Operator's responsibility is creation of
Redpanda configuration that has correctly registered advertised
Kafka API and assign correct unique and unused node port. With the
help of Kubernetes Service type node port the unused port can be
injected into container host port. This service doesn't have selector,
so that no rule should be created in `iptable` or any other CNI
implementation. Init container will retrieve from the Kubernetes API
server information about node and its public IP. That container will
then change the Redpanda configuration, so that internal and external
ports and addresses will be in Kafka API and Advertised Kafka API
section.

## Drawbacks

The nodes must be exposed to some other networks. This scope goes
beyond the Kubernetes. There is no easy way for the operator to
manage firewalls that can be blocking the network traffic.

## Rationale and Alternatives

The more natural way of integrating with k8s resource is to use service
type `LoadBalancer` per pod. This would add latency and cost will scale
as the bytes comes through.

Each instance could be registered in DNS provider of one's choice.
The benefits with this approach is nice looking brokers names. On the
other hand the latency of propagating change in DNS systems could
not be acceptable.

Following `external-dns`
[tutorial for headless services](https://github.com/Kubernetes-sigs/external-dns/blob/eaf933328f4f7ed1db1546145fa81ed5a61cae3b/docs/tutorials/hostport.md)
there should be an easy way of exposing each pod. During the tests
in AWS EKS the DNS A records shows internal IP address of a Pod in
the AWS route 53 service.

Exposing all brokers via LoadBalancer and Envoy proxy. The Envoy can
reroute connection to proper broker. The Envoy can be replaced with
any other proxy Nginx etc. There will be bottleneck in hot path as
every connection will go through LB + proxy. It doesn't scale well
as one client can starve the network hardware.

## Unresolved questions

- What do to do with the migration of the brokers to different machine?
- How this feature integrates with TLS?
  - How the subject alternative names should look like?
  - Does the advertised Kafka API need to be change from
    HOST_PUBLIC_IP:PORT to DNS_NAME:PORT?

---

### Rfcs 20220119 Raft Replication Simplification (`docs/rfcs/20220119_raft_replication_simplification.md`)

- Feature Name: Raft replication simplification - phase one
- Status: in-progress
- Start Date: 2022-01-19
- Authors: Michal Maslanka
- Issue: #1829 (Quality initiative)

## Current solution

Redpanda's Raft protocol implementation currently supports two replication 
modes, or consistency levels - the Quorum replication mode and the 
leader ack mode.

The Quorum replication mode works the same as the 'original' 
version of the Raft protocol, i.e. the replication call finishes when the requested entry 
is safely replicated to the majority of the replicas and is flushed. 

In the leader ack mode (Kafka's ACKS=1) the `replicate` call finishes as soon as the write 
to the leader disk is finished (before it is flushed). 

The entries' visibility for  readers is also influenced by 
the replicate consistency levels. 
For entries replicated with Quorum consistency level, Redpanda makes them visible as soon as 
they are fsynced on the majority of nodes. On the other hand, entries with the 
`leader_ack` consistency level are made visible as soon as they 
are written to the follower's disk, but they are not required to be fsynced.

In the current implementation, the two consistency levels
use totally different approaches and code paths. Quorum level replication 
requests go through the batching on the leader `raft::replicate_batcher`
and then they are dispatched to the followers via `raft::replicate_entries_stm`. 
Raft `replicate_stm` writes data to the leader disk and then dispatches 
`append_entries_requests` to the follower in parallel with the leader disk flush.
In this case, Raft implementation may have multiple requests in flight per follower. 
(currently the number of requests pending is limited by per follower semaphore). 

Leader ack consistency replication is implemented differently. 
Control is returned to the caller immediately after the data is written 
to the underlying `storage::log`. On the next heartbeat, the follower 
is recognized to be behind the leader and is recovered by `raft::recovery_stm`. 
The `recovery_stm` reads a chunk of data and sends it to the follower 
in a loop (a single request at a time).

## Purpose of the simplification

There are multiple reasons that makes the simplification necessary and demanded, 
the most important are:

- code simplification
- unification of approaches
- better handling of requests with mixed consistency levels (currently they cause latency spikes)

Also some future features that will benefit from simplification are

- in-memory acks
- leaders that are remote to data 


## Proposed solution

Proposed solution assumes that both currently supported consistency levels will 
be handled in the same way with use of `replicate_batcher` and `replicate_entries` stm. 
This approach has several advantages over the one currently 
used for `leader_ack` consistency level. 

The current implementation of `quorum_acks` replication is robust and pretty 
well tested. No large code changes are required to use the 
`replicate_batcher` and `replicate_stm` to handle `leader_acks` replication. 
The way that the `replicate_stm` handles `append_entries` requests dispatching 
and leader log writes will allow us to make them parallel in future. 

Additionally, using `replicate_stm` prevents us from creating the 
append entries request twice since batches coming from `replicate_batcher` 
are not first handed off to the disk and then read again, but rather shared to 
generate a leader disk write and follower `append_entries` requests. 
This approach significantly reduces the `storage::batch_cache` pressure and may 
also reduce the disk usage (in some clusters with limited memory cache, the hit 
ratio is not always 1.0 even if there are no reads other than the reads issued by 
`replicate_stm`).

### Backpressure

In the current `quorum_ack` replication implementation, the backpressure is 
propagated from the followers to the `raft::replicate` caller 
(the Kafka Produce request handler effectively). Currently, the backpressure 
propagation is based on the per-follower semaphore since the RPC layer 
lacks a backpressure propagation mechanism. 
When `replicate_entries` stm is unable to acquire follower dispatch units, 
it will wait, holding the `replicate_batcher` memory reservation lock so that 
it can not accept writes indefinitely. When the `replicate_batcher` memory 
reservation semaphore is exhausted, it prevents the first stage of `replicate_stages` 
from finishing, not allowing more requests to be processed. 

The current `leader_ack` mechanism lacks backpressure propagation. 
The leader can be an arbitrary large number of entries ahead of followers. 
This solution may be a problem since the user has an impression that data is 
replicated while the data is actually stored only on the leader.

In the proposed solution, we are going to handle backpressure in the same way 
for both consistency levels. This way there will be a guarantee that followers  
can always keep up with the leader.


### Memory usage

Memory usage control will be achieved in the same way it is done right now, 
i.e. we will not finish first stage of replication when the `replicate_batcher` 
memory semaphore is exhausted. This way we will prevent the Kafka layer from 
buffering an indefinite number of batches in pending replicate requests. Replicate 
batcher memory units are released after the request is handed off to the 
socket so that all the user space buffer can be freed. 
In future, we may consider a per-shard semaphore that controls the total amount 
of memory used by all the replicate batchers.

### Recovery STM coordination - finishing recovery

`recovery_stm` is started every time the leader identifies the follower as 
being behind. Since recovery stm is currently used for both recovering the 
follower (when it is actually behind) and delivering follower data replicated 
with the `leader_ack` consistency, the `recovery_stm` implementation contains some 
optimization to minimize the `leader_ack` latency. 

There are two main optimizations:
- not stopping the recovery when the last replicate request consistency level is `leader_ack`
- triggering the next dispatch loop immediately after the leader disk append

These optimizations make the handling of the recovering follower more complex in 
other scenarios, such as `leadership_transfer`.

Since there is no coordination between the two paths that the follower append 
entry requests are sent from (i.e. `replicate_stm` and `recovery_stm`), those 
requests may be reordered and cause log truncations and redeliveries, which lead 
to increased latency.

In the proposed solution, since we are not longer going to rely on the 
recovery STM to deliver messages to the follower in normal conditions, the complicated
handling of recovery finishing logic will no longer be needed. 
When the follower is up to date with the leader, `recovery_stm` is simply not running.

The only situation that will require coordination is when the follower 
is no longer behind the leader, but `replicate_stm` skips dispatching requests 
to the follower as it is still recovering. This may lead to a situation in 
which the next leader append will happen before the `recovery_stm` received the follower 
response for the last request, i.e. it cannot finish recovery because the log end already 
advanced. This case may be handled by coordinating dispatching follower 
requests from `recovery_stm` and `replicate_stm`. 
We can introduce the last sent offset to 
`raft::follower_index_metadata` to check whether recovery should finish 
instead of the check being based only on the follower response.

This way, the `replicate_stm` will not skip sending a request to the follower if 
it is the next request the follower is supposed to receive. On the other hand, the 
`recovery_stm` will not try to send the same request again, but will finish, 
assuming the response will be successful.

### Implementation details

To implement the simplified solution, we will pass the `consistency_level` 
parameter to the `raft::replicate_batcher::replicate` method. 
The `raft::replicate_entries_stm::apply_on_leader` method will return immediately after 
the successful `leader_append`. The full quorum replication round will be finished 
after the `raft::replicate_entries_stm::wait_for_majority()` returns. 
The replicate batcher, based on the cached request `consistency_level`, will decide when 
to release the request, either after `apply_on_leader` finishes or 
`wait_for_majority` finishes.

As in current implementation, the `raft::replicate_entries_stm` will
release batcher memory units after dispatching the follower request. 
This will allow propagating backpressure for both consistency levels.

---

### Rfcs 20220207 Connections Per Ip (`docs/rfcs/20220207_connections_per_ip.md`)

- Feature Name: Client conection limits
- Status: implemented
- Start Date: 2022-02-07
- Authors: John Spray
- Issue: 

# Executive Summary

Add per-client state to track and limit the number of connections open.

Similar to KIP-308, adapted to Redpanda's thread per core model.

## What is being proposed

New configuration properties:
- kafka_connections_max (optional integer, unset by default)
- kafka_connections_max_per_ip (optional integer, unset by default)
- kafka_connections_max_overrides (list of strings, default empty)

New prometheus metric:
- vectorized_kafka_rpc_connections_rejected

When a connection is accepted which exceeds one of the configured
bounds, it is dropped before reading or writing anything to the socket.

## Motivation

* Avoid hitting system resource limits (e.g. open file handles)
  related to port counts
* Prevent rogue clients from consuming unbounded memory (we allocate
  some userspace buffer space to all connections, and the kernel
  allocates some buffers too).
* Help users notice if a client is acting strangely: an application
  author opening large numbers of connections will tend to notice
  themselves when they hit the limit, rather than the operator
  of the cluster noticing when the cluster performance is impacted.
* Provide a client allow-listing/deny-listing mechanism (e.g.
  kafka_max.connections_per_ip is set to zero and
  kafka_max_connections_per_ip_overrides specifies
  permitted clients)
* Continuity of experience for Kafka users accustomed to
  max.connections.per.ip

## How

### Accepting a connection

If both configuration properties are unset, then no action is taken: this
preserves the existing unlimited behaviour.

A sharded service storing per-client tokens for open connections, clients are
mapped to shards by hash of their IP.

In kafka::protocol::apply, acquire a token at start and relinquish it at end,
probably using a RAII structure, although this function already provides a
convenient localized place to take+release a resource explicitly.

On first connection from a client, the shard servicing the connection makes a
cross-shard call to the core that holds state for this client IP hash.  This
initializes the client's state:

```c++
struct client_conn_quota {
  // The max per connection.  This is set by applying
  // the combination of per_ip and per_ip_override settings.
  uint32_t total;

  // How many connections are currently open.  This may be at
  // most `total`, although it may exceed it transiently if
  // configuration properties are decreased during operation.
  uint32_t in_use;
}
```

The connection quota state is stored in a absl::btree_map: since the overall
number of unique clients maybe high (so a flat store is risky), but the entries
are small (so a std::map would generate a lot of tiny allocations).

### Optimization: caching tokens

In the case where the client is not close to exceeding its limits, we can avoid
the overhead of a cross-core call by caching some per-client tokens on each
shard.

This cache can be pre-primed on the first client connection, if the total
tokens for the client is greater than the core count: that way, subsequent
connections from the same client will usually be accepted without the need for
a cross-core call.

### Handling a configuration change

Changes to the configuration properties affect all outstanding
client_conn_quota objects: the `total` attribute must be recalculated.

If `total` is decreased, then it is also necessary to call out to all other
shards and revoke any cached tokens that are held in excess of the new limit.

## Impact

When this feature is enabled, a per-connection latency overhead is added, as
the shard handling the connection must dispatch a call to the shard that owns
the client IP state.  A cross core call can take up to a few microseconds,
although the impact on P99 latency can be worse if the target core is very busy.

On a system where the connection count limit is significantly larger than the
core count, this overhead will be avoided in most cases, as each core will
store some locally cached tokens to allow them to accept connections.


# Guide-level explanation

## How do we teach this?

* Presume reader knows what a TCP connection is
* Explain how Kafka clients map to TCP connections: that a producer or consumer
  usually opens a connection and uses it over a long period of time, but
  a program that constructs many Kafka clients in parallel might open more
  connections.  The relationship between clients and connections is not 1:1,
  as clients may e.g. be both a producer and a consumer, and the client may
  open additonal connections for reading metadata.
* Explain why it's bad to have an unbounded number of connections from a
  client:
  * Each connection has some memory overhead that might exhaust redpanda memory
  * Each connection counts against OS-level limits (`ulimit`)
  * It's probably not what the user intended for efficient operation, as the
    Kafka protocol generally re-uses a smaller number of connections.
* Introduce the new configuration properties, including the syntax of the
  `overrides` property
  * Mention the caveat that decreasing the limits does not terminate any
    currently open connections.

## Interaction with other features

* Dependency on centralized config for live updates

## Drawbacks

### Latency overhead on first message in a new connection:

Mitigating factors:
* Sensible workloads generally operate on established connections
* No overhead if feature is not enabled.

## Alternatives Considered

### Do nothing

An application may exhaust server resources by opening a very large number of
connections (e.g. by instantiating many Kafka clients).  This can be mitigated
in environments where the application design is tightly controlled and the
server cluster is sized to match, but in real-life environments this is
unlikely to be the case & it is relatively easy for a naively-written application
to open unbounded numbers of Kafka client connections.  For example, consider
a web service that opens a Kafka connection for each request it serves, and
does not have a limit on its own concurrent request count.

### Total connection count, rather than per-client-IP

To protect redpanda from resource exhaustion, it is not necessary to
discriminate between clients: we could just have a single overall connection
count.

This prevents redpanda from having resources exhausted, but does not help
preserve availability of the system: one misbehaving server can crowd out
other clients.

### Shard-local limits

We could avoid some complexity and latency by setting a per-core connection
limit, rather than a total connection limit across all cores.  This would
enable all cores to apply their limits using local state only.

This is reasonable at small core counts, but tends to fall down at larger core
counts: consider a 128 core system, where a per-core limit to tolerate
reasonable applications (e.g. 10 connections) would result in a total limit of
1280, much higher than the user wanted.

### Per-user limits

In many environments, the client IP is not a meaningful discriminator between
workloads -- workloads are more likely to be identifiable by their service
account username.

Strictly speaking, it is not possible to limit connections by username, because
we have to accept a connection and receive some bytes to learn the username, but
it could still be useful to drop connections immediately after authentication
if a user has too many connections already.

Adding per-user limits in future is not ruled out, but the scope of this RFC is
limited to the more basic IP-based limits, to get parity with KIP-308.

### Kernel-space (eBPF/BPF/iptables) enforcement

Where the connection count for a particular source IP has been exhausted, and/or
was set to zero, we could more efficiently drop connections by configuring the
kernel to drop them for us, rather than doing it in userspace.  This kind of
efficiency is important in filtering systems that guard against DDoS situations.

The connection count limits in this RFC are primarily meant to protect against
misbehaving (but benign) applications rather than active attack, so the overhead
of enforcing in userspace is acceptable.  Avoiding kernel mechanisms also helps
with portability (including to older enterprise linux distros) & potential
permissions issues in locked-down environments that might forbid userspace
processes from using some kernel space mechanisms.

All that said, extending our userspace enforcement with lower level mechanisms
in future could be a worthwhile optimization.  In Kubernetes environments,
we would need to consider whether to implement this type of enforcement within
Redpanda pods, or within the Kubernetes network stack where external IPs are
exposed.

---

### Rfcs 20220208 Admin Api Basic Auth (`docs/rfcs/20220208_admin_api_basic_auth.md`)

- Feature Name: HTTP Basic Auth for admin API
- Status: implemented.
- Start Date: 2022-02-08
- Authors: John Spray
- Issue: https://github.com/redpanda-data/redpanda/issues/3702

# Executive Summary

Enable admin API clients to authenticate using SASL username/password credientials
passed into an HTTP "Basic" Authentication header. 

These checks will be off by default for backward compatibility, but enabled during
setup of new clusters.

## What is being proposed

A cluster configuration property:
* `admin_api_require_auth` (boolean, default false).

When this property is true, requests to the admin API will be rejected unless they
carry a valid username+password, and that username is the existing `superusers` list.

## Why (short reason)

Currently, users' only secure option for the admin API is to enable mTLS and
disable plain HTTP access, and then only issue mTLS client certs to users who
are permitted to perform administrative operations.

This causes two problems:
- Users who have not set up mTLS (and may not want to) have no authentication
  mechanism for the admin API.  They have to either refrain from using it entirely,
  or accept the security risk of having an unauthenticated listener (e.g. on 127.0.0.1:9644)
- Users who _have_ set up mTLS have no way of differentiating regular kafka clients
  from a privileged identity that should be able to perform administrative operations.

Resolving these issues unlocks use of the Admin API in 
production environments, including new centralized configuration workflows that
rely on admin API access.

## How (short plan)

The existing `superusers` configuration property will be used to discriminate between
regular users, and users that should be permitted to use the admin API.

The plaintext password in the Authentication header can be validated against stored
scram credentials, by passing the password through the same salt/hmac steps that we would use
if setting a new password.


# Guide-level explanation

## How do we teach this?

- Explain use of `--user` and `--password` on `rpk`.
- Explain that the credentials can be the same as a Kafka (SASL) username
- Explain concept of a superuser, and how the list of superusers can be
  edited like other configuration properties (i.e. via `rpk cluster config edit`)

## What if the user locks themselves out?

In extreme circumstances, they can disable auth using 
`rpk cluster config force-reset admin_api_require_auth` while Redpanda is offline.

However, unless the user forgets a password they should not be able to lock themselves
out: the API endpoint for modifying cluster config will ensure that when auth is enabled,
it is always enabled by an authenticated user who is already a superuser, so that a naive
user can't permanently lock themselves out just by enabling auth before they've declared
themselves a superuser.

## Introducing new named concepts.

All the concepts in use here already exist:
- HTTP Basic Authentication is a standard
- The concept of a superuser already exists, we are extending its definition
  to include access to the admin API.
- The concept of a SASL user account already exists.

# Reference-level explanation

## Detailed design

### Auth hooks in admin API

We are using seastar's built-in HTTP framework, which does not provide
generic request/response hooks: there is not a place to connect an authentication
callback.

As a substitute, we will introduce a helper wrapper around our method callbacks,
that applies authentication rules before running the inner function.

### Runtime configuration property changes

To support dynamic updates to `superusers` and `admin_api_require_sasl`, the
values of these properties will simply not be cached anywhere.  Reading them
out of the per-shard cluster configuration object at the start of each request
is a cheap operation.

### Which endpoints will be authenticated?

All endpoints will be be subject to authentication rules, with two exceptions:
- `/v1/metrics` - the metrics are not generally sensitive, and it is very
  convenient to be able to point prometheus at an endpoint without having
  to arrange secrets
- `/v1/status` - a node should be willing to tell any client whether it is up.

## Interaction with other features

All `rpk` commands that rely on admin API access will now require proper
`--user` and `--password` flags on systems where `admin_require_sasl` is true.

The kubernetes operator currently relies on unauthenticated access to the
admin API.   It also creates and stores superuser credentials when creating
a cluster, so extending it to use these credentials to access the API is straightforward,
as long as the end user does not themselves modify these superuser credentials.

During kubernetes cluster creation, the operator can bootstrap authentication as follows:
- Bring up nodes in an initial state with no external admin API connectivity, only
  exposing it on the pod-local IPs.
- Using admin API via pod local IPs, create superuser account
- Using admin API via pod local IPs, set `admin_api_require_sasl` to true.
- Add external admin API listeners and restart nodes.

Clearly the above sequence is simplified if the cluster does not have any external
connectivity to the admin API.

## Upgrades

SCRAM credentials may be either sha256 or sha512 type, whereas Authentication header does
not specify which to use.  For existing systems, we can only validate passwords by trying
both algorithms.

The credential store should be extended to record the algorithm for each credential.  This
might be done as part of implementing this feature, or we might choose to defer that until
the `serde` transition, for convenient backward compatible encoding.

## Telemetry & Observability

* WARN-level log messages on access denied events
* WARN-level log message on startup if an unauthenticated listener is running

## Drawbacks

* rpk command lines are made much more verbose with user/password information.
  * This is mitigated in cloud environments where the user is provided
    with an `rpk.sh` wrapper that automatically loads superuser credentials from
    kubernetes.
  * Bare metal customers are likely to do something similar.
* Risk of user "locking themselves out of the house" once the admin API is hidden
  behind the same credentials that the admin API is used to edit.
  * This is intrinsic to improving security
  * We will provide a recovery procedure for this case.
* Risk of user locking out the kubernetes operator, for example by deleting
  the superuser account that the operator uses.
  * This risk already existed in general, as the operator does not have
    a special privilege mechanism to protect its own credentials from the end user
  * If this occurs, SREs will need to use the same mechanisms as if the user
    had locked themselves out.
* Bare-metal users might store passwords in cleartext scripts
  * This is still less bad than the API being entirely unauthenticated.
  * If users have their own secrets infrastructure, they should be using it
    to store these secrets somewhere safe.

## Rationale and Alternatives

### Alternative: do nothing

### Alternative: HTTP Digest auth

This is not possible using existing stored SCRAM credentials, because
HTTP digest hashing is different from the hashing used in SCRAM credentials.

We could implement this for newly created credentials if we calculated
and stored the HTTP Digest MD5 hash of the password during account creation
(at which time, we have the plaintext password).

While HTTP digest auth is slightly better than basic auth in that it doesn't
transmit passwords in the clear, it has many security issues of its own, and
in practice either of these protocols should be transported over TLS for
a meaningful level of security.

### Alternative: restrict admin API to unauthenticated localhost listener

If the admin API only listened on 127.0.0.1, then users could SSH to a redpanda
node and drive admin operations from there.  This has several downsides:
- Security is still weak, as any process running on the redpanda node can
  access the admin API
- It is inconvenient to do all operations via an SSH session
- The admin API can't issue a redirect to another node (e.g. to redirect
  to controller leader) if all the nodes are only listening on localhost.

### Alternative: create separate admin API credentials (e.g. API keys)

It is convenient but not essential to use the same system of credentials
for the admin API that we use for Kafka protocol.  We could implement
a separate pool of API keys.

This approach would increase the surface area of the feature substantially,
as we would not be able to use the existing user CRUD commands, and would
be more complex to explain to the user compared with the existing user/password
framework they know from the world of Kafka.

### Alternative: shared secret in node config

For use cases like the kubernetes operator, it might be sufficient to simply
store a random string in the node configuration file (redpanda.yaml), and
use this as a shared secret.

This has several downsides:
* We would be storing a de-facto cleartext password
* Key changes would require node restart
* User would be unable to distinguish clients (for audit purposes) by
  user account if all shared a statically set token for admin API access.

### Alternative: only authenticate writes

For some use cases (e.g. monitoring), it might be convenient to give the user read-only
access without providing a secret.

However, this cannot be applied uniformly to all endpoints, for example the config
endpoint includes sensitive configuration properties like the cloud access key.  It would
also remove the ability to record an audit record of which users were accessing what
information.

### Alternative: per-endpoint RBAC

This isn't so much an alternative as a possible extension.  The authorization layer
can be extended in many ways in the future: this RFC is just about getting some
kind of authentication (SASL user account) combined with the most basic level
of authorization (the superuser list).

## Future work

Integration with operator, if enable admin API auth in the cloud is needed.  Currently
admin API is not exposed to users on FMC clusters.

---

### Rfcs 20220209 Leader Epoch Support Rfc (`docs/rfcs/20220209_leader_epoch_support_rfc.md`)

- Feature Name: Leader epoch handling in Redpanda
- Status: in-progress
- Start Date: 2022-02-09
- Authors: Michal Maslanka
- Issue: https://github.com/redpanda-data/redpanda/issues/3689

# Executive summary

In Kafka, some of the client requests are fenced with the leader epoch. 
The leader epoch is a 32-bit integer that is increased every time the partition 
leadership changes. The leader epoch is used to detect stale metadata, 
fence produce and fetch requests, and recognize truncations.

## What is being proposed

The proposal is to use the Raft protocol `term` as a Kafka leader epoch. The `term` is semantically 
identical to Kafka's leader epoch as it is increased every time the Raft group 
leadership changes and it stays constant when leader is active.

## Why (short reason)

The leader epoch is used by Kafka clients to detect truncations and stale metadata. 
In Redpanda we are missing that mechanism. This may lead to some issues that 
can occur when one of the nodes is isolated and does not receive metadata updates. 
Returning the leader epoch will allow the client to identify this situation 
and update metadata against a different broker. 

## How (short plan)

We will expose Raft group terms that are already gathered in the Redpanda metadata 
infrastructure as leader epochs. Additionally, we will implement the 
`OffsetForLeaderEpoch` Kafka API to provide full compatibility. When exposed to 
the Kafka layer `raft::term` type will be wrapped in `kafka::leader_epoch` helper.


# Motivation
## Why are we doing this?

Adding leader epoch support will allow us to better handle stale metadata 
and unclean truncations.

### Note on unclean truncations in redpanda

Unclean leader election may happen in Kafka when stale replica is allowed to 
become a leader. This may lead to situation in which partition 
high watermark would move backward. 

Since Redpanda replication is based on Raft, we do not allow for unclean leader 
election that may lead to the high watermark being decreased. The only situation in 
which the high watermark may be decreased is simultaneous failure of the majority of 
nodes handling relaxed consistency writes such as `ACKS=1` and `ACKS=0`. It is 
more likely to happen for topics with replication factor of `1`. Then it would 
be enough for only one node to fail to move high watermark backward.

## What use cases does it support?

We will add support for a leader epoch in each request that requires it and is 
currently handled by Redpanda.

## What is the expected outcome?

The validation of a leader epoch that is consistent with Kafka.

# Guide-level explanation
## How do we teach this?

This is a standard feature of the Kafka protocol and we may use existing, related KIPS.

- KIP-101 Alter Replication Protocol to use Leader Epoch rather than High Watermark for Truncation
- KIP-232 Detect outdated metadata using leaderEpoch and partitionEpoch
- KIP-320 Allow fetchers to detect and handle log truncation
- KIP-359 Verify leader epoch in produce requests (not yet implemented in Kafka)

# Reference-level explanation
## Detailed design - What needs to change to get there

### Leader epoch metadata

We need to store the last stable leader term in `cluster::partition_leaders_table`.
The term will be directly mapped to the leader epoch returned in `MetadataResponse`. 

We will introduce a type representing the current leader and the term. 

```c++
    struct leader_term {
        std::optional<model::node_id> leader;
        model::term_id term;
    };

```

### Offset for leader epoch API handler

The `OffsetForLeaderEpoch` API returns the last offset in the request leader epoch plus 
one. That is the first offset of the next leader epoch, i.e. an offset 
on which the requested leader epoch ends.

The `OffsetForLeaderEpoch` is handled only by the requested partition leader. 
In order to implement that feature, we need to add an API that will allow us to 
query for last offset in the given term. Then the offset will have to be translated 
with an offset translator to obtain a valid Kafka offset.

For local partitions this is an easy task since segments are stored in a sorted 
collection. Each segment can contain only single term. 

For remote partitions (cloud storage), we will use 
information stored in partition manifests. The manifest contains the segment 
term and the offsets stored in a segment. Given that we do not allow two 
terms to be stored in the same segment, we may use that information to determine 
last term offset without downloading actual data.
## Drawbacks

We add more complexity to the implementation of few Kafka APIs. The complexity 
however will allow us to better recover from failures and network partitions and
will provide better compatibility with Kafka APIs. 

## Rationale and Alternatives

One of the considered alternative was to use a separate integer tracking 
leadership changes. That would give us a separation of concerns between term 
and leader epoch. This approach was rejected since the introduced integer would 
be semantically identical to already existing `term` and implementing all the 
necessary pieces would require a lot of work and coordination between all 
redpanda submodules.

---

### Rfcs 20220209 Limit Connections Rate (`docs/rfcs/20220209_limit_connections_rate.md`)

- Feature Name: Ability to limit connection creation rate per shard
- Status: in-progress
- Start Date: 2022-02-09
- Authors: Vadim Plakhtinskiy
- Issue: 

# Executive Summary

- Set a limit on the rate with which the shard accepts new connections
- Set a limit on connection creation rate per IP


Similar to KIP-612, adapted to Redpanda's thread per core model

## What is being proposed

New configuration properties (For start we choose global config for connection rate):
- kafka_max_connection_creation_rate (optional integer, default unset)
- kafka_max_connection_creation_rate_ovverides (string, default "")


When a connection is accepted which exceeds one of the configured
bounds, it is blocking before reading or writing anything to the socket.


## Motivation

* Avoid hitting system resource limits (for example CPU)
* Avoid geting connections storms
* Allow broker doing useful work instead of handling high connection creation rate

## How

If both configuration properties are unset, then no action is taken: this
preserves the existing unlimited behavior.
 
We will track connection rate on each shard independently (without cross-core communication).
 
For limit connections rate we will use a token bucket algo. Each bucket is semaphore for counting connections in the current second. For overrides we will use a hash table from ip -> semaphore.
 
struct connection_rate {
   ss::semaphore current_rate
   ss::lower_clock last_update_time
}
 
For new connection we will do:
1. check ss::lowres_clock::now() and last_update_time, if second was changed we should signal to increase tokens to kafka_max_connection_creation_rate
2. run wait(timeout, 1) for semaphore

### How to add new tokens to semaphore
Problem: what if we have the maximum number of connections and should block new connections
waiting for the next second, who will signal them to unblock?
We should avoid this situation, and will spawn background fiber if available_units is 0 after lock.
Background fiber will signal(kafka_max_connection_creation_rate) on next second after spawn if needed (if zero available tokens)
 
We can run background fiber, which will add kafka_max_connection_creation_rate tokens to semaphore on each second.
 
Note: each signal(kafka_max_connection_creation_rate) also should update last_update_time

### Integration with current network code
We have net::server for different type tcp connection (rpc_kafka, internal_rpc, etc)
 
Idea is adding new settings to server_configuration (settings for connection rate).
On server::start it will init internal structure with connection_rate info,
 
In server::accept before
 
ssx::spawn_with_gate(_conn_gate, [this, conn]() mutable {
  return apply_proto(_proto.get(), resources(this, conn));
});
            
We will check the current connection rate.
 
If we should spawn background fiber (to increase tokens in semaphore) we will spawn it before
 
ssx::spawn_with_gate(_conn_gate, [this, conn]() mutable {
  return apply_proto(_proto.get(), resources(this, conn));
});

### Handling a configuration change

Changes to the configuration properties affect all outstanding
connection_rate objects: the `kafka_max_connection_creation_rate` attribute must be recalculated.

### Metrics
We should reprot about current rate on each shard

## Impact

1. Each connection will store start connection time
2. Use clock to comapre times, but we will use ss::lower_clock
3. Background fiber to update rate limit in semaphore

# Guide-level explanation

## How do we teach this?
* Presume reader knows what a TCP connection and connections rate is
* Explain why it's bad to have an unbounded number of connections rate
  * Each connection has some memory overhead that might exhaust redpanda memory
  * Connections storm can increase CPU and can create problem with CPU and performance problems
* Introduce the new configuration properties, including the syntax of the
 `overrides` property
  * Mention the caveat that decreasing the limits does not terminate or stopping any
   currently open connections.


## Interaction with other features

* Dependency on centralized config for live updates

# Drawbacks

The lowest effective rate the user can set is multiplied by the number of shards.
So user should think about it

## Alternatives Considered

### Do nothing

A client may exhaust server resources by opening a very large number of
connections in one second. This is okay as long as all clients are well-behaved and the
system is carefully sized for its workload, neither of which are reliably true
in real life deployments.

### Run backround fiber
We can run background fiber each second to refresh tokens count in semaphore.
In thins case we will avoid getting and comparing time for each connection,
and only fiber will run signal()

### Node limits (not shard)

It will add more compexity to share info about current rate to each core.
Also we should sharding info about ovverides and map each ip to core for saving
info about rate per ip

---

### Rfcs 20221018 Cluster Bootstrap (`docs/rfcs/20221018_cluster_bootstrap.md`)

- Feature Name: Cluster Bootstrap Revamp
- Status: implementation in progress
- Start Date: 2022-10-18
- Authors: Andrew Wong
- Issue: #333, #2793


# **RFC: Cluster Bootstrap**

## **Summary**

This document describes upcoming changes to the steps performed by Redpanda when instantiating a cluster, with the intent to simplify configuration of nodes and eliminate the class of incidents caused by the incorrect configuration of node IDs and seed servers. The user-facing effect of this proposal is that the configuration of each node can be made identical with one another -- we will require no special node and no per-node specifics when configuring a cluster.


## **Background**


TL;DR



* We currently require that there is one node with an empty `seed_servers` config, and this heterogeneous configuration requirement causes trouble when not implemented
* A single server (the “root” node) instantiates the cluster as a cluster of size one; all other nodes join the cluster by routing RPCs to the “seed servers”, who eventually route the RPCs to an existing cluster (typically the root node when first bootstrapping); during this period, the cluster is up and running, but is arguably in a surprising state (one node cluster)
* We require manual assignment of node IDs, which is annoying to configure and easy to get wrong, especially in cloud deployments
* It is unpleasant to bootstrap superusers today via environment variables
* There is no way to start a cluster with license

Current implementation of cluster formation:



* **Root:** becomes leader immediately, begins servicing RPCs, processes join requests and runs Raft member add operation when other servers attempt to join.
* **Seeds:** tries to connect to all seed servers to try to join. Also receives requests to join from non-seed servers, and redirects the requests if we’ve joined the cluster and we’re not the leader.
* **Non-seeds:** tries to connect to all seed servers to try to join the cluster.

The above requirements of today’s configuration has led to poor experience for customers by way of incidents as severe as having a controller topic with split brain.


## **Requirements Overview**



* Make it significantly more difficult to misconfigure a cluster in a way that starts multiple clusters or no clusters, especially in the face of a wiped node (or several).
* To that end, simplify configuration by allowing admins to configure the nodes with identical config files (within reason, users likely still want to advertise different addresses per node) — no more root servers, and no more mandatory hard coding of node IDs.
* Until a cluster has been fully started, the nodes of the cluster should perform no action that may depend on the initial cluster state.
    * Calls to the admin server API or Kafka API should return a retriable error code.
    * We should not nag because of a missing license, nor should we perform any action that depends on having a valid license.


## **Design**


### **High level design**

Since there are several issues being addressed in this project, there are a few disparate tasks involved in this overhaul of bootstrapping:



* Drop the requirement that each node configuration must include a node ID. Users may still supply node IDs, but as a part of bootstrapping a cluster, Redpanda will enforce that newly added nodes will have unique node IDs. Nodes will assign themselves a UUID (persisted in their kv-store), and register themselves with the controller leader, who assigns a unique node ID. Reusing a node ID that has already been used will be strictly disallowed.
* Remove the period of cluster formation where the cluster only has a single node. Instead, clusters will start with at least three (configurable) nodes, instantiating a controller Raft group with the seed servers list, and growing the cluster as nodes start up.
* Initial cluster configuration will be done via either environment variables, or bootstrap config file, depending on sensitivity of the configs (e.g. user credentials are secrets, licenses aren’t)
* Optionally gate the starting of a cluster on the run of a new CLI command. Admins can provide initial user credentials and a license to this command. Before the successful completion of this command, no node in the cluster will respond to Kafka requests, admin requests, etc. This waiting is more of a cherry-on-top as a more sure-fire gate to cluster formation accidents – the above three design points get us most of the way there in solving customer pain points, and will be the focus for Q3.


### **Multi-node cluster formation**

Rather than relying on a single node to be the root node, the entire seed servers list will be taken to be the initial Raft group for the controller. Once all the seed servers are up, each seed server should determine whether or not all seed servers have the same list of seed servers. If so, the nodes are primed for starting a cluster and should await instruction to do so. While waiting for start up, we may initialize most subsystems, but must not initialize the controller, a bulk of the admin server, Kafka API, to name a few.

The initialization on seed servers (non-seed servers continue with today’s join-until-success logic), in short:



* **Seeds:** tries to connect to all seed servers to ensure they all think the list of seed servers, and other cluster configs, is identical. Once this is verified, and no cluster UUID already exists, awaits user input to start its Raft instances with the seed servers list as the initial Raft group.
* **Non-seeds:** tries to connect to all seed servers to try to join, sending a UUID along with a node ID. No strong requirements on the seed servers list here, other than pointing to a member of the cluster.
* **Seeds after cluster formation:** as indicated by the presence of a non-empty controller log or a cluster UUID, follows the above non-seeds logic.

The initial seed server check will be done via a new RPC endpoint with the following request and response structure:


```
struct initial_cluster_info_request {
    // Empty.
};

struct initial_cluster_info_reply {
    // A cluster UUID, if one already exists, indicating we should join.
    string cluster_uuid;

    // Manually assigned node ID, if any, so we can form our initial mapping of UUIDs.
    int node_id;

    // UUID of the destination node, so we can form our initial mapping of UUIDs.
    string node_uuid;

    // Seeds from the destination server. Maybe useful to log if there's a problem.
    // Useful in detecting the empty_seed_servers_forms_cluster=True case.
    vector<string> seed_servers;
};
```


The subsystem that serves these RPCs must initialize before the controller forms its initial Raft group. As such, we will instantiate an RPC protocol on the seed servers only that accepts the `initial_cluster_info` calls. Once the controller has initialized, we will add the normal RPC protocol for normal runtime.

Alternate approaches to initial controller group sizing



* Start with size three; taking the first three nodes in the seed server list to be the founding controller Raft group, ensuring identical configs across each config.
    * **Downside:** will likely take longer to run through several config changes than initialize a large Raft group.
    * **Downside:** having a policy on how to pick three nodes seems fragile and difficult to evolve.
    * **Downside:** unintuitive and surprising to have only a subset of the seed servers participate in bootstrapping, and leaves room for error and races when considering nodes starting up at different rates.
* Allow waiting for a specific cluster size before starting the controller group, rather than inferring based on seed servers list. Initialize the cluster only once the expected size is reached, e.g. propagating a node list via gossip, and only starting a Raft group of that size when prompted by the init command after the gossip state indicates the cluster has the nodes expected.
    * **Upside:** some kind of gossip mechanism can be used as a foundational piece for other interesting features
    * **Downside:** added complexity, maintenance burden, support burden, of yet another distributed protocol subsystem
    * **Side note:** the proposal here can be a piece to work towards a gossip-based initialization. Such an implementation would wait for the gossip subsystem to have all nodes agree on a founding set of nodes, as well as an initial assignment of node IDs for them, after which, cluster initialization can proceed.

Alternate approaches to controller group initialization



* Rather than waiting for an explicit init command, unconditionally proceed once we know all seed servers are healthy and configured uniformly.
    * **Upside:** deployment is easier to automate because there is no special step to run init.
    * **Downside:** we no longer have an explicit step with which to provide initial cluster-wide configs.
    * **Downside:** this is still error prone in the case of all seed servers being wiped and restarted, e.g. the seed servers are haphazardly placed in the same region and there is a region failure. While more difficult, it’s still possible to restart some nodes empty and start up with two clusters (compared to the current proposal, if the seeds start empty, they would still need to await the init command, rather than start a new cluster).
* Have the default for `wait_for_init` to be false
    * **Downside:** as a default, this would be unsafe for the reasons stated above.

Alternate approaches to `initial_cluster_info` endpoint



* Use an entirely separate server bound to a different port
    * **Upside:** likely better tested approach
    * **Downside:** may require additional configuration
* Use configs for the RPC server, but once we’re done, stop the initial_cluster_info server, and then start a new server with the runtime protocol
    * **Downside:** there may be a brief blip where no RPC server is unavailable
    * **Downside:** unclear whether this approach is any better tested


### **Single-node cluster formation**

It doesn’t seem unreasonable that some clusters knowingly start out with cluster size 1, and then grow as use cases grow more mature on the cluster. Additionally, the existing user experience for developer clusters relies heavily on being able to not provide a seed server list.

To that end, we will continue to support supplying an empty seed servers list or a seed server list of size one, which will be construed to mean starting a cluster with size 1 node.

Alternatives



* Don’t support single node clusters period.
    * **Downsides:** use case doesn’t sound unreasonable, especially for folks just testing out a binary. A goal of this project is to make trying out Redpanda easy and failsafe, and single-node/local clusters seem like an important part of that experience.
* Force users to specify when they want a single node cluster by setting some `minimum_initial_cluster_size` configuration


### **Controller log initialization**

A new controller log command will be added to indicate the initial desired state of the cluster, including initial admin secrets, license, etc. This will be a new controller command with roughly the following payload:


```
struct cluster_init_cmd_data {
    string cluster_uuid;
    vector<tuple<uuid, node_id, addr>> seed_servers;
    vector<pair<credential_user, scram_credential>> user_creds;
    string license;
};
```


Upon electing the leader of the controller group, the leader should determine whether it has this message by checking its kv-store for a cluster UUID. If not, it will replicate a new message. The contents of the message must be known before controller group formation.

Only once this entry is replicated and applied to a node will that node open up its admin server, Kafka ports, etc. and continue with bootstrapping. The apply of this message must also affect the corresponding state in the members table, feature table, and security manager.

It should be noted that clusters today already initialize a cluster UUID as a part of the metrics subsystem. This old cluster UUID is used for phone home.

Alternatives



* Rather than a single message, rely on existing RPCs and send multiple messages on the controller log.
    * **Downsides:** seems unrealistic to be able to do this robustly; messages would not be atomic and would thus be fragile against lag, restarts, etc.
    * **Upsides:** reuse existing services.


### **Node ID assignment**

Upon starting up, if there is not already a UUID in the local kv-store, initialize a UUID and persist it in the local key-value. This will be the node’s UUID.

The `join_node` request messages already include an unused `node_uuid` field and an existing `node_id` field (moving forward, just a hint). The response `node_id` field will indicate what `node_id` has been assigned.

Attempts to join a cluster with the same `node_uuid` after having already joined that cluster with that UUID should fail. This would indicate a bug wherein a node that has data attempts to join a cluster twice.

Attempts to join a cluster with an in-use `node_id` should fail, if the UUID in the request differs from the one persisted in the members table.

Auto-generated `node_ids` will be monotonically increasing for now, even if decommissioning and recommissioning a node — such a node will be assigned a new `node_uuid` and a new `node_id`.

Today, nodes only send join requests after it has started up its RPC server and is ready to begin accepting Raft requests for its node ID. In the future, when nodes start up without a node ID, we need to uphold this ordering to avoid some extraneous delays in bootstrapping. To that end, assignment of node ID and joining a cluster will be two separate operations. The controller provides a means to perform operations serially and in a deterministic order, so it seems like an ideal place to assign node IDs.

When starting up with no node ID, it will send a join request to the seed servers to replicate a controller command that registers the input UUID with an unused node ID. Since the ordering of controller apply operations is deterministic, the IDs assigned are deterministic as well, and can be inferred by the order in the log with a simple counter.

Initial node IDs for the seed servers are taken to be the index in the seed servers list.

Alternatives to node ID assignment mechanism



* Don’t infer node IDs based on order, and instead use some kind of leadership-based lock to assign a node ID before replicating to followers. This is similar to how the ID allocator works.
    * **Upside:** easier to debug from just the controller log
    * **Downside:** more complicated implementation, and allows for gaps in the node ID space (e.g. in case leadership is lost and then gained again on the same node while registering a UUID)

Alternatives to initial node ID assignment.



* Initial node IDs are taken as input to the init command, like as a map from {`seed_server_addr`=>`seed_server_node_id`}.
    * **Downside:** not much benefit here, if an admin wanted to explicitly set node ID, they should do that in the node configs

Alternatives to mechanics of node ID assignment before join



* Extend Raft config change with node UUIDs
    * **Downside:** hard to uphold the invariant that we join only once we’re ready to receive Raft RPCs
    * **Downside:** likely results in more complex code, since we would be overloading the purpose of Raft config change
    * **Downside:** only the controller group would need to care about UUIDs – all other Raft groups can continue using node IDs, so making a fundamental change to the underlying Raft implementation seems like overkill
* Add an additional field to `configuration_update` that indicates new UUIDs
    * **Downside:** seems redundant with `replicas_to_add` field
    * **Upside:** no changes to foundational Raft data types

**Superuser configuration**

Continue using environment variables, and eventually implement as a part of an initialize command.

Alternatives



* Supply as a node config
    * **Downside:** this cannot be the only solution to this problem because not all deployment modes have dedicated secret storage (k8s does, on-prem probably doesn’t).
    * **Downside:** muddies the distinction between how we initialize node configs (i.e. through yaml), and how we set cluster configs (i.e. through centralized configs in the controller). Maybe this isn’t so bad.
    * **Downside:** potential to misconfigure, since these would be set on each node, and could end up being set incorrectly.


### **License configuration**

NOTE: as of v22.3, this is not yet implemented.

Supply as a part of the bootstrap config YAML, and eventually implement as a part of an initialize command.

Alternatives



* Supply via init command
* Supply as an environment variable
    * **Downside:** not a great experience for on-prem
* Supply as a node config
    * **Downside:** muddies the distinction between how we initialize node configs (i.e. through yaml), and how we set cluster configs (i.e. through centralized configs in the controller). Maybe this isn’t so bad.
    * **Downside:** potential to misconfigure, since these would be set on each node, and could end up being set incorrectly.
* Supply as an argument to the Redpanda binary
    * **Upside:** it’s a step towards not requiring two-step bootstrap altogether
    * **Downside:** needs to be included on every node


### **What should a single-node cluster look like?**

Single node clusters will not be a special case. Everything else about the cluster will still reflect the changes laid out in this RFC. Cluster UUIDs will still be assigned, node UUIDs will still be generated, joining nodes will be automatically assigned node IDs, etc.

Alternatives



* Have the behavior be identical to how it was before
    * **Downside:** would incur lots of legacy code that is difficult to wade through
    * **Downside:** knowingly unsafe
* Add a configuration `empty_seed_servers_starts_cluster` that means seed servers should attempt to join if receiving `initial_cluster_info` response with empty `seed_servers`
    * **Upside:** allows users to not change their configs
    * **Downside:** knowingly unsafe


### **Init command**

NOTE: as of v22.3, this is not yet implemented.

Only once the rest of this project is done will we consider implementing an init command.

The `rpk init` tool will trigger the request to start the cluster, and will return only after the cluster has formed successfully, returning its cluster UUID.

The tool will take as input just a single address, potentially localhost. If the node is a seed server and is awaiting initialization, it will run and forward to all other seed servers. If the node is not a seed server, it will just forward the request via RPC to each seed server. If a cluster has already been formed (as indicated by a persisted cluster UUID), the server will return immediately with a cluster UUID.

The tool will periodically retry the request until a cluster UUID is returned, at which point it will exit.

Alternatives



* No init command, and instead provide all configs (except initial user credentials) as arguments to `rpk cluster start`
    * **Upside:** significantly easier to deploy in an automated way (no need to have a one-off command that runs once)
    * **Downside:** vulnerable in the case where several nodes go down and come up completely empty (i.e. all seed servers are in a single zone, a zone failure wiped out data in that zone, all seed servers then try to form a single new cluster).


### **Security implications of init command**

NOTE: as of v22.3, this is not yet implemented.

Either run from localhost, or run with a superuser defined in the bootstrap YAML and `RP_BOOTSTRAP_USER`.


### **Automated deployment considerations**

Since we are still going to support the existing pattern of configuration, we won’t have to immediately update any k8s operator, ansible, etc. deployments. Some already have mechanisms to avoid starting up with an empty seed servers list after initial bootstrap.

Moving forward, at the very least, we should remove the generation of node ID from these deployments.

Longer term, we should be very intentional about the seed servers list, e.g. at least one seed in each AZ, to avoid creating a new cluster when an AZ goes down.


### **Reconciling existing cluster UUID with new cluster UUID**

NOTE: as of v22.3, this is not yet implemented.

There already exists a cluster UUID that is inferred by every node, as it is based on the controller log. We hash the first controller messages and timestamps, and use that as the seed for the cluster UUID, deterministically ensuring it’s identical on every node. This gets persisted in the config manager.

Moving forward, the metrics reporter should check to see if a cluster ID exists in the config, and pull it directly from the shard-local config if so. If not, it should use the cluster UUID.

---

--- METRICS ---
- Files Extracted: 96
- Estimated Token Budget: ~84859 tokens
- Canonical Reference: https://codewiki.google/github.com/redpanda-data/redpanda