Пишем высокопроизводительный rate-limiter на golang
О чём статья
Недавно я познакомился с golang и в качестве практики решил реализовать алгоритмы rate limiting'а из статьи c arxiv. В этой статье мы рассмотрим написание rate-limiter'а в виде отдельно стоящего сервиса и в виде middleware, используя in-memory кэш или Redis, добавим поддержку REST и gRPC, подключим Prometheus и поймём как добиться атомарности операций с минимальными задержками.
Кому полезно
Новичкам в golang или же людям, которые по какой-то причине также решили использовать собственные реализации rate-limiter'ов заместо уже существующих решений.
Предисловие
Указанная ранее статья покрывает 3 алгоритма rate-limiting'а:
Token bucket
Fixed window
Rolling window
Суть данных алгоритмов вы можете прочитать самостоятельно, я же сфокусируюсь на том, как написать данные алгоритмы таким образом, чтобы rate-limiter работал быстро и надёжно.
Конфигурация
Первое, с чего стоит начать, это конфигурация вашего limiter'а. В своей реализации я ставил следующие требования:
Возможность переключаться между алгоритмами
Поддержка in-memory кэша и Redis
Метрики в Prometheus
Поддержка REST и gRPC
Из этих требований вытекает следующий файл конфигурации:
store: "redis"
use_algo: "token_bucket"
metrics:
use_prometheus: true
port: 9091
algo_settings:
capacity: 10
rate: 0.1
backends:
in_memory:
default_ttl: 30 # seconds
eviction_time: 120 # seconds
shards_count: 256
redis:
addr: "redis:6379"
password: "m1re1di77sss"
db: 0
default_ttl: 1000sourcecraft.dev
Рассмотрим его подробнее:
store отвечает за используемое хранилище, может быть "redis" или же "in_memory"
use_algo отвечает за используемый алгоритм
metrics позволяет включать и отключать поддержку выгрузки и подсчёта метрик во время работы, а также за порт, на котором позже будет поднят мультиплексер для метрик
algo_settings для каждого алгоритма разные, отвечает за настройки, которые будут использоваться алгоритмом
backends - настройки каждого из хранилищ, в in_memory default_ttl отвечает за ttl ключей в хранилище, eviction_time отвечает за частоту очистки in-memory хранилища от протухших ключей, shard_count отвечает за то, на сколько частей in-memory кэш будет шардирован, число должно быть степенью 2, позже я объясню почему
Конфигурация для middleware выглядит похожим образом:
type InMemoryBackendSettings struct {
DefaultTtl int64 yaml:"default_ttl"
EvictionTime int64 yaml:"eviction_time"
ShardsCount uint32 yaml:"shards_count"
}
type RedisBackendSettings struct {
Addr string yaml:"addr"
Password string yaml:"password"
Db int yaml:"db"
DefaultTtl int64 yaml:"default_ttl"
}
type AlgoSettings struct {
Capacity *float64 yaml:"capacity"
Rate *float64 yaml:"rate"
WindowSize *int64 yaml:"window_size"
MaxRequests *int64 yaml:"max_requests"
}sourcecraft.dev
Бэкенды для limiter'а
Думаю, ни для кого не секрет, что Redis умеет выполнять Lua скрипты атомарно, что является идеальной функциональной возможностью для надёжного rate-limiting'а. В таком случае наш бэкенд limiter'а также должен уметь выполнять операции атомарно, отсюда вытекает интерфейс бэкенда:
type InMemoryStoreInterface interface {
Update(key string, newFn func() any, fn func(v any) error) error
Close() error
}sourcecraft.dev
Update принимает на вход ключ, по которому мы хотим принять решение, а также 2 колбэка:
newFn - функция создания нового объекта в хранилище, в случае если данный ключ в хранилище ещё не появлялся. Это позволяет избежать создания данного объекта в случае, если ключ уже содержится в хранилище. Конечно, можно было сначала проверять и только потом добавлять, но это добавляет лишней логики клиенту, использующему хранилище.
fn - собственно функция, содержащая логику limiting'а
In-memory кэш
Наше хранилище под капотом будет использовать мапу, во избежание race condition'ов мы будем использовать mutex. Но ставить mutex на всю мапу слишком дорого, так как другие клиенты всё ещё должны пользоваться хранилищем.
Отсюда вытекает необходимость в шардировании. Пока одна горутина выполняет логику по ключу x, мы всё ещё можем работать с ключом y:
type shard struct {
mu sync.Mutex
entries map[string]*cacheEntry
}
type cacheEntry struct {
entry any
lastAccess int64
}
type ShardedInMemoryStore struct {
shards []*shard
shardsCount uint32
expirationTime int64
ticker *time.Ticker
done chan any
once sync.Once
}
func NewCache(
expirationTime int64,
evictionInterval time.Duration,
shardCount uint32,
) *ShardedInMemoryStore {
ticker := time.NewTicker(evictionInterval)
cache := &ShardedInMemoryStore{
shards: make([]*shard, shardCount),
shardsCount: shardCount,
expirationTime: expirationTime,
done: make(chan any),
ticker: ticker,
}
for i := range shardCount {
cache.shards[i] = &shard{entries: make(map[string]*cacheEntry)}
}
go func() {
defer ticker.Stop()
for {
select {
case <-cache.done:
return
case <-ticker.C:
cache.handleEviction()
}
}
}()
return cache
}
sourcecraft.dev
Также удаление протухших ключей теперь не будет лочить всю мапу, а только один конкретный шард:
func (cache *ShardedInMemoryStore) handleEviction() {
currentTime := time.Now().Unix()
for _, shard := range cache.shards {
shard.mu.Lock()
for key, value := range shard.entries {
if currentTime-cache.expirationTime > value.lastAccess {
delete(shard.entries, key)
}
}
shard.mu.Unlock()
}
}sourcecraft.dev
А также ранее упомянутый Update:
func (cache *ShardedInMemoryStore) Update(key string, createFn func() any, fn func(v any) error) error {
shard := cache.getShard(key)
shard.mu.Lock()
defer shard.mu.Unlock()
now := time.Now().Unix()
value, present := shard.entries[key]
if !present {
value = &cacheEntry{entry: createFn(), lastAccess: now}
shard.entries[key] = value
}
value.lastAccess = now
return fn(value.entry)
}sourcecraft.dev
Именно здесь реализуется столь необходимая атомарность, мы лочим mutex на шарде и выполняем операцию, не блокируя другие шарды и не допуская race condition'ов.
Так почему количество шардов должно быть степенью двойки?
Для того, чтобы распределять клиентские ключи более-менее равномерно по шардам нужно использоват функцию хэширования, которая не только будет раскладывать ключи равномерно, но и будет быстрой. Под такую функцию подходит Fowler–Noll–Vo. А чтобы после этого быстро замапить хэш на шард можно воспользоваться трюком. Вкратце можно очень быстро искать остатки от деления на степень двойки просто используя формулу: hash & (shardsCount - 1):
func (cache *ShardedInMemoryStore) getShard(key string) *shard {
h := uint32(2166136261)
for i := 0; i < len(key); i++ {
h ^= uint32(key[i])
h *= 16777619
}
return cache.shards[h&(cache.shardsCount-1)]
}sourcecraft.dev
Redis кэш
С Redis всё проще:
package cache
import (
"context"
"github.com/redis/go-redis/v9"
)
type RedisStoreInterface interface {
RunLuaScript(ctx context.Context, script *redis.Script, keys []string, args ...any) (bool, int64, error)
DefaultTtl() int64
}sourcecraft.dev
Передаём контекст, скрипт и массив ключей, в данном случае массив ключей это требование библиотеки для работы с Redis в Golang. Массив всегда будет состоять из одного элемента.
DefaultTtl - это getter, который используется для token bucket алгоритма, чтобы удалением ключа при использовании Redis занимался сам Redis, а не приложение.
package cache
import (
"context"
"fmt"
"github.com/redis/go-redis/v9"
)
type RedisStore struct {
client *redis.Client
defaultTtl int64
}
func NewRedisStore(client *redis.Client, defaultTtl int64) *RedisStore {
return &RedisStore{client: client, defaultTtl: defaultTtl}
}
// RunLuaScript runs a Lua script that returns {allowed(0|1), retryAfter}.
func (r *RedisStore) RunLuaScript(
ctx context.Context,
script *redis.Script,
keys []string,
args ...any,
) (bool, int64, error) {
res, err := script.Run(ctx, r.client, keys, args...).Int64Slice()
if err != nil {
return false, 0, err
}
if len(res) != 2 {
return false, 0, fmt.Errorf("unexpected length of script output")
}
return res[0] == 1, res[1], nil
}
func (r *RedisStore) DefaultTtl() int64 {
return r.defaultTtl
}sourcecraft.dev
Используем созданные хранилища
Теперь рассмотрим функцию запуска limiter'а:
package main
import (
"context"
"fmt"
"log"
"net"
"net/http"
"os"
"os/signal"
"qwrttqr-rate-limiter/core/config"
"qwrttqr-rate-limiter/core/server/algos"
"qwrttqr-rate-limiter/core/server/facade"
"qwrttqr-rate-limiter/core/server/limiting"
"qwrttqr-rate-limiter/core/server/metrics"
ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1"
"strconv"
"syscall"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
"google.golang.org/grpc"
)
func main() {
if err := run(); err != nil {
log.Fatal(err)
}
}
func run() error {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
cfg, err := config.ReadConfig()
if err != nil {
return fmt.Errorf("failed to read config: %w", err)
}
rl, err := limiting.NewRateLimiter(cfg)
if err != nil {
return fmt.Errorf("create rate limiter: %w", err)
}
defer rl.Close()
var limiter algos.Limiter = rl
errCh := make(chan error, 3)
shutdownMetrics := func(ctx context.Context) {}
if cfg.Metrics.UsePrometheus {
reg := prometheus.NewRegistry()
rec := metrics.NewPromRecorder(reg)
limiter = metrics.Instrument(limiter, rec)
metricsMux := http.NewServeMux()
metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second}
go func() { errCh <- metricsSrv.ListenAndServe() }()
log.Printf("metrics available on %s", metricsSrv.Addr)
shutdownMetrics = func(ctx context.Context) {
_ = metricsSrv.Shutdown(ctx)
}
}
limitingMux := http.NewServeMux()
limitingMux.HandleFunc("GET /health", facade.HealthHandler)
limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter))
limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second}
grpcSrv := grpc.NewServer()
ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter))
tcpLis, err := net.Listen("tcp", ":8081")
if err != nil {
return fmt.Errorf("listen grpc: %w", err)
}
go func() { errCh <- limitingSrv.ListenAndServe() }()
log.Printf("listening http on %s", limitingSrv.Addr)
go func() { errCh <- grpcSrv.Serve(tcpLis) }()
log.Printf("listening grpc on %s", tcpLis.Addr())
select {
case err := <-errCh:
return err
case <-ctx.Done():
}
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
grpcDone := make(chan struct{})
go func() { grpcSrv.GracefulStop(); close(grpcDone) }()
select {
case <-grpcDone:
case <-shutdownCtx.Done():
grpcSrv.Stop()
}
shutdownMetrics(shutdownCtx)
return limitingSrv.Shutdown(shutdownCtx)
}
sourcecraft.dev
Давайте по порядку:
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()sourcecraft.dev
Создаём shutdown context, который срабатывает на нажатие Ctrl-C либо при получении syscall.SIGTERM (сигнал, посылаемый родительским процессом на завершение нашего процесса, например когда тушим докер-контейнер). defer stop() нужен для того, чтобы прекратить обрабатывать данные сигналы, сразу перед возвратом из run.
cfg, err := config.ReadConfig()
if err != nil {
return fmt.Errorf("failed to read config: %w", err)
}
rl, err := limiting.NewRateLimiter(cfg)
if err != nil {
return fmt.Errorf("create rate limiter: %w", err)
}
defer rl.Close()sourcecraft.dev
Далее читаем конфиг и создаём сам limiter, у limiter'а есть функция Close, нужная для закрытия соединения с Redis. в in-memory кэше она просто закрывает канал, что останавливает удаление ключей из кэша.
var limiter algos.Limiter = rl
errCh := make(chan error, 3)sourcecraft.dev
Следующие строки позволяют создать указатель с типом интерфейса на limiter:
package algos
import (
"context"
)
type Request struct {
Key string
RequiredTokens float64
}
type Decision struct {
Allowed bool
// RetryAfter is seconds
RetryAfter int64
}
type Limiter interface {
Allow(ctx context.Context, req Request) (Decision, error)
}sourcecraft.dev
Так как NewRateLimiter возвращает не интерфейс, а указатель на структуру, поддерживающую Close и Allow:
type RateLimiter struct {
algos.Limiter
io.Closer
}
func NewRateLimiter(cfg config.StandaloneServiceConfiguration) (_ *RateLimiter, err error) {
var (
mem cache.InMemoryStoreInterface
rdb *redis.Client
closer io.Closer
)
switch cfg.Store {
case "in_memory":
cacheInstance := cache.NewCache(
cfg.Backends.InMemory.DefaultTtl,
time.Duration(cfg.Backends.InMemory.EvictionTime)*time.Second,
cfg.Backends.InMemory.ShardsCount,
)
mem, closer = cacheInstance, cacheInstance
case "redis":
redisClient := redis.NewClient(&redis.Options{
Addr: cfg.Backends.Redis.Addr,
Password: cfg.Backends.Redis.Password,
DB: cfg.Backends.Redis.Db,
})
rdb, closer = redisClient, redisClient
default:
return nil, fmt.Errorf("unsupported store: %q", cfg.Store)
}
defer func() {
if err != nil {
_ = closer.Close()
}
}()
var limiter algos.Limiter
switch cfg.UsedAlgo {
case "token_bucket":
if err := algos.ValidateTokenBucketConfig(cfg.AlgoSettings); err != nil {
return nil, err
}
store, err := algos.NewTokenBucketStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl)
if err != nil {
return nil, err
}
limiter = &algos.TokenBucketLimiter{
Capacity: *cfg.AlgoSettings.Capacity,
Rate: *cfg.AlgoSettings.Rate,
Store: store,
Now: realClock,
}
case "fixed_window":
if err := algos.ValidateFixedWindowConfig(cfg.AlgoSettings); err != nil {
return nil, err
}
store, err := algos.NewFixedWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl)
if err != nil {
return nil, err
}
limiter = &algos.FixedWindowLimiter{
MaxRequests: *cfg.AlgoSettings.MaxRequests,
WindowSize: *cfg.AlgoSettings.WindowSize,
Store: store,
Now: realClock,
}
case "rolling_window":
if err := algos.ValidateRollingWindowConfig(cfg.AlgoSettings); err != nil {
return nil, err
}
store, err := algos.NewRollingWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl)
if err != nil {
return nil, err
}
limiter = &algos.RollingWindowLimiter{
MaxRequests: *cfg.AlgoSettings.MaxRequests,
WindowSize: *cfg.AlgoSettings.WindowSize,
Store: store,
Now: realClock,
}
default:
return nil, fmt.Errorf("unsupported rate limiting algorithm: %s", cfg.UsedAlgo)
}
return &RateLimiter{Limiter: limiter, Closer: closer}, nil
}sourcecraft.dev
Возвращаясь обратно в run:
shutdownMetrics := func(ctx context.Context) {}
if cfg.Metrics.UsePrometheus {
reg := prometheus.NewRegistry()
rec := metrics.NewPromRecorder(reg)
limiter = metrics.Instrument(limiter, rec)
metricsMux := http.NewServeMux()
metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second}
go func() { errCh <- metricsSrv.ListenAndServe() }()
log.Printf("metrics available on %s", metricsSrv.Addr)
shutdownMetrics = func(ctx context.Context) {
_ = metricsSrv.Shutdown(ctx)
}
}sourcecraft.dev
Мы помечали rl, как структуру, реализующую интерфейс как раз ради поддержки метрик здесь. О том, как метрики работают я расскажу позже, сейчас достаточно понимания того, что мы создаём специальный HTTP сервер, с которого Prometheus будет забирать метрики. Для подсчёта данных метрик мы используем Registry, а в функции instrument мы оборачиваем limiter в мидлварь, которая будет обновлять счётчики.
Функция shutdownMetrics обновляется в зависимости от того, включаем ли у себя метрики или нет, это нужно, чтобы сделать graceful shutdown данного сервера.
limitingMux := http.NewServeMux()
limitingMux.HandleFunc("GET /health", facade.HealthHandler)
limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter))
limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second}
grpcSrv := grpc.NewServer()
ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter))
tcpLis, err := net.Listen("tcp", ":8081")
if err != nil {
return fmt.Errorf("listen grpc: %w", err)
}
go func() { errCh <- limitingSrv.ListenAndServe() }()
log.Printf("listening http on %s", limitingSrv.Addr)
go func() { errCh <- grpcSrv.Serve(tcpLis) }()
log.Printf("listening grpc on %s", tcpLis.Addr())
select {
case err := <-errCh:
return err
case <-ctx.Done():
}sourcecraft.dev
После этого создаём HTTP и gRPC сервера и ждём в select на ранее созданном NotifyContext, до момента пока не придёт сигнал на завершение процесса (os.Interrupt, syscall.SIGTERM) или ошибка.
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
grpcDone := make(chan struct{})
go func() { grpcSrv.GracefulStop(); close(grpcDone) }()
select {
case <-grpcDone:
case <-shutdownCtx.Done():
grpcSrv.Stop()
}
shutdownMetrics(shutdownCtx)
return limitingSrv.Shutdown(shutdownCtx)sourcecraft.dev
В случае, если к нам пришла ошибка или же сигнал на завершение процесса, создаём контекст с таймаутом и просим все наши сервера выключиться. В особенности это касается gRPC сервера, он самостоятельно поддерживает GracefulStop, но если он не успеет сделать это за 10 секунд, мы выключим его принудительно.
Используем ранее созданные бэкенды
Помните, что мы создали Redis и InMemory бэкенды для наших алгоритмов, теперь можно посмотреть, как мы пользуемся ими, рассмотрим пример с Rolling Window:
type RollingWindowStore interface {
CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error)
}
type RollingWindowLimiter struct {
WindowSize int64
MaxRequests int64
Store RollingWindowStore
Now func() int64
}
func (rwl *RollingWindowLimiter) Configure() {
if rwl.Now == nil {
rwl.Now = func() int64 {
return time.Now().Unix()
}
}
}
func ValidateRollingWindowConfig(cfg config.AlgoSettings) error {
if *cfg.MaxRequests <= 0 {
return fmt.Errorf("the max requests param should be strictly greater then 0")
}
return CheckRequiredFields(cfg, []string{"window_size", "max_requests"})
}
type InMemoryRollingWindowStore struct {
Backend cache.InMemoryStoreInterface
}
type RedisRollingWindowStore struct {
Backend cache.RedisStoreInterface
}
func NewRollingWindowStore(store string, cacheInstance cache.InMemoryStoreInterface, redisClient *redis.Client, redisDefaultKeyTTL int64) (RollingWindowStore, error) {
switch store {
case "in_memory":
return &InMemoryRollingWindowStore{Backend: cacheInstance}, nil
case "redis":
return &RedisRollingWindowStore{Backend: cache.NewRedisStore(redisClient, redisDefaultKeyTTL)}, nil
default:
return nil, fmt.Errorf("unsupported store: %s", store)
}
}sourcecraft.dev
Ещё один интерфейс(RollingWindowStore), нужен для генерализации работы с бэкендом, таким образом, сам алгоритм будет просто вызывать функцию CheckAndIncrement, передавая параметры, а бэкенд будет выполнять её в атомарном варианте.
func (rwl *RollingWindowLimiter) Configure() {
if rwl.Now == nil {
rwl.Now = func() int64 {
return time.Now().Unix()
}
}
}sourcecraft.dev
Данная функция нужна для тестов, так как в тестах придётся мокать функцию получения текущего времени.
type RollingWindowState struct {
Timestamps []int64
}
func (rws *RollingWindowState) check(
windowSize,
maxRequests,
now int64,
) (bool, int64) {
windowStart := now - windowSize
i := 0
for ; i < len(rws.Timestamps); i++ {
if rws.Timestamps[i] > windowStart {
break
}
}
if i > 0 {
copy(rws.Timestamps, rws.Timestamps[i:])
rws.Timestamps = rws.Timestamps[:len(rws.Timestamps)-i]
}
if int64(len(rws.Timestamps)) < maxRequests {
rws.Timestamps = append(rws.Timestamps, now)
return true, 0
}
retryAfter := rws.Timestamps[0] - windowStart
return false, retryAfter
}sourcecraft.dev
Сам алгоритм устроен несложно, мы просто отслеживаем, какие временные метки уже вышли за временное окно, и если оставшихся меток меньше, чем лимит запросов, клиент может выполнить очередной запрос. Иначе нужно ответить, через какое количество секунд освободится хотя бы одно место.
Далее подвязываем данную функцию к бэкенду:
func (store *InMemoryRollingWindowStore) CheckAndIncrement(
_ context.Context,
key string,
windowSize,
maxRequests,
now int64,
) (allowed bool, retryAfter int64, err error) {
err = store.Backend.Update(
key,
func() any { return &RollingWindowState{Timestamps: make([]int64, 0, 8)} },
func(v any) error {
st, ok := v.(*RollingWindowState)
if !ok {
return fmt.Errorf("unexpected state type %T for key %q", v, key)
}
allowed, retryAfter = st.check(windowSize, maxRequests, now)
return nil
})
return allowed, retryAfter, err
}sourcecraft.dev
То же самое в Redis:
var rollingWindowScript = redis.NewScript(`
local requestId = ARGV[1]
local windowSize = tonumber(ARGV[2])
local maxRequests = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local windowStart = now - windowSize
redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", windowStart)
local count = redis.call("ZCARD", KEYS[1])
if count < maxRequests then
redis.call("ZADD", KEYS[1], now, requestId)
redis.call("EXPIRE", KEYS[1], windowSize)
return {1, 0}
else
local oldest = redis.call("ZRANGE", KEYS[1], 0, 0, "WITHSCORES")
local retryAfter = 0
if oldest and oldest[2] then
retryAfter = tonumber(oldest[2]) - windowStart
end
return {0, retryAfter}
end
`)
func (s *RedisRollingWindowStore) CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error) {
requestId := uuid.New().String()
return s.Backend.RunLuaScript(ctx, rollingWindowScript, []string{key}, requestId, windowSize, maxRequests, now)
}sourcecraft.dev
В Redis реализации стоит обратить внимание на то, что мы генерируемrequestId, всё потому, что мы храним наши элементы в множестве, а в множестве могут храниться только уникальные элементы. Но как тогда быть в ситуации, когда 2 и более запросов прилетают в одну секунду? Для этого и создан requestId, он добавляет уникальность каждому запросу, поэтому если в одну секунду прилетит 1000 запросов, они все посчитаются как разные запросы.
А что с метриками?
Как я ранее говорил, rate-limiter поддерживает подсчёт метрик. Начнём с функции instrument, о которой говорилось ранее:
package metrics
import (
"context"
"qwrttqr-rate-limiter/core/server/algos"
"time"
)
type instrumented struct {
next algos.Limiter
rec Recorder
}
func Instrument(next algos.Limiter, rec Recorder) algos.Limiter {
return &instrumented{next: next, rec: rec}
}
func (i *instrumented) Allow(ctx context.Context, req algos.Request) (algos.Decision, error) {
start := time.Now()
dec, err := i.next.Allow(ctx, req)
elapsed := time.Since(start)
switch {
case err != nil:
i.rec.ObserveDecision(Errored, elapsed)
case dec.Allowed:
i.rec.ObserveDecision(Allowed, elapsed)
default:
i.rec.ObserveDecision(Denied, elapsed)
}
return dec, err
}sourcecraft.dev
Данная функция просто возвращает структуру, которая обладает методом Allow, таким образом мы подменяем наш оригинальный limiter обёрткой, которая поддерживает подсчёт метрик.
Recorder реализован так:
package metrics
import "time"
type Result uint8
const (
Allowed Result = iota
Denied
Errored
NumResults
)
type Recorder interface {
ObserveDecision(r Result, elapsed time.Duration)
}sourcecraft.dev
И конкретная реализация данного интерфейса под Prometheus выглядит так:
package metrics
import (
"time"
"github.com/prometheus/client_golang/prometheus"
)
type PromRecorder struct {
decisions [NumResults]prometheus.Counter
durationHist [NumResults]prometheus.Observer
}
func NewPromRecorder(reg prometheus.Registerer) *PromRecorder {
counterVec := prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "ratelimiter",
Name: "decisions_total",
Help: "Each decision count by its type.",
}, []string{"result"})
reqDurHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
Namespace: "ratelimiter",
Name: "decision_duration_seconds",
Help: "Execution time of one decision in seconds.",
Buckets: []float64{0.00005, 0.0001, 0.0005, 0.001, 0.1, 0.5, 1, 2},
}, []string{"result"})
reg.MustRegister(counterVec)
reg.MustRegister(reqDurHistVec)
p := &PromRecorder{}
p.decisions[Allowed] = counterVec.WithLabelValues("allowed")
p.decisions[Denied] = counterVec.WithLabelValues("denied")
p.decisions[Errored] = counterVec.WithLabelValues("errored")
p.durationHist[Allowed] = reqDurHistVec.WithLabelValues("allowed")
p.durationHist[Denied] = reqDurHistVec.WithLabelValues("denied")
p.durationHist[Errored] = reqDurHistVec.WithLabelValues("errored")
return p
}
func (p *PromRecorder) ObserveDecision(r Result, dur time.Duration) {
p.decisions[r].Inc()
p.durationHist[r].Observe(dur.Seconds())
}sourcecraft.dev
Здесь мы создаём счётчики и гистограммы под каждый результат rate-limiter'а и реализуем ObserveDecision, который их обновляет.
gRPC и REST
gRPC
Пишем .proto файл:
syntax = "proto3";
package ratelimiter.v1;
option go_package = "qwrttqr-rate-limiter/gen/ratelimiter/v1;ratelimiterv1";
import "google/protobuf/duration.proto";
service RateLimiter {
rpc Allow(AllowRequest) returns (AllowResponse);
}
message AllowRequest {
string key = 1;
double tokens = 2;
}
message AllowResponse {
bool allowed = 1;
google.protobuf.Duration retryAfter = 2;
}sourcecraft.dev
Вызываем protoc и остаётся просто реализовать полученные интерфейсы:
protoc -I proto --go_out=. --go_opt=module=qwrttqr-rate-limiter --go-grpc_out=. --go-grpc_opt=module=qwrttqr-rate-limiter ratelimiter/v1/ratelimiter.protosourcecraft.dev
package facade
import (
"context"
"qwrttqr-rate-limiter/core/server/algos"
ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1"
"time"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"
)
type GRPCServer struct {
ratelimiterv1.UnimplementedRateLimiterServer
l algos.Limiter
}
func NewGRPCServer(l algos.Limiter) *GRPCServer {
return &GRPCServer{l: l}
}
func (s *GRPCServer) Allow(ctx context.Context, in *ratelimiterv1.AllowRequest) (*ratelimiterv1.AllowResponse, error) {
tokens := in.GetTokens()
if tokens == 0.0 {
tokens = 1.0
}
req, err := NewRequest(in.GetKey(), tokens)
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
d, err := s.l.Allow(ctx, req)
if err != nil {
return nil, status.Error(codes.Unavailable, "rate limiter error")
}
return &ratelimiterv1.AllowResponse{
Allowed: d.Allowed,
RetryAfter: durationpb.New(time.Duration(d.RetryAfter) * time.Second),
}, nil
}sourcecraft.dev
REST
Для REST всё совсем просто:
package facade
import (
"net/http"
"qwrttqr-rate-limiter/core/server/algos"
"strconv"
)
func HealthHandler(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte("ok"))
}
func LimitHandler(l algos.Limiter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
key := r.Header.Get("X-Client-Key")
tokens := 1.0
if s := r.Header.Get("X-Required-Tokens"); s != "" {
n, err := strconv.ParseFloat(s, 64)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
tokens = n
}
req, err := NewRequest(key, tokens)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
d, err := l.Allow(r.Context(), req)
switch {
case err != nil:
http.Error(w, "rate limiter error", http.StatusInternalServerError) // store failure
case !d.Allowed:
w.Header().Set("Retry-After", strconv.FormatInt(d.RetryAfter, 10))
http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
}
}
}sourcecraft.dev
Middleware
Для реализации поддержки middleware можно использовать такую схему:
Приложение создаёт себе бэкенд, используя Redis или оставаясь на in-memory кэше
Используя созданный бэкенд создаём себе rate-limiter, с нужным алгоритмом
package middleware
import (
"qwrttqr-rate-limiter/core/config"
"qwrttqr-rate-limiter/core/server/algos"
"qwrttqr-rate-limiter/core/server/cache"
"time"
"github.com/redis/go-redis/v9"
)
var realClock = func() int64 { return time.Now().Unix() }
type Backend interface {
TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error)
FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error)
RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error)
}
type memoryBackend struct {
mem cache.InMemoryStoreInterface
}
func NewInMemoryBackend(cfg config.InMemoryBackendSettings) Backend {
return &memoryBackend{mem: cache.NewCache(
cfg.DefaultTtl,
time.Duration(cfg.EvictionTime)*time.Second,
cfg.ShardsCount),
}
}
type redisBackend struct {
redis cache.RedisStoreInterface
}
func NewRedisBackend(client *redis.Client, redisDefaultKeyTTL int64) Backend {
return &redisBackend{redis: cache.NewRedisStore(client, redisDefaultKeyTTL)}
}
func newTokenBucket(cfg config.AlgoSettings, store algos.TokenBucketStore) (algos.Limiter, error) {
if err := algos.ValidateTokenBucketConfig(cfg); err != nil {
return nil, err
}
return &algos.TokenBucketLimiter{
Capacity: *cfg.Capacity, Rate: *cfg.Rate, Store: store, Now: realClock,
}, nil
}
func newFixedWindow(cfg config.AlgoSettings, store algos.FixedWindowStore) (algos.Limiter, error) {
if err := algos.ValidateFixedWindowConfig(cfg); err != nil {
return nil, err
}
return &algos.FixedWindowLimiter{
WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock,
}, nil
}
func newRollingWindow(cfg config.AlgoSettings, store algos.RollingWindowStore) (algos.Limiter, error) {
if err := algos.ValidateRollingWindowConfig(cfg); err != nil {
return nil, err
}
return &algos.RollingWindowLimiter{
WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock,
}, nil
}
func (b *memoryBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) {
return newTokenBucket(cfg, &algos.InMemoryTokenBucketStore{Backend: b.mem})
}
func (b *memoryBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newFixedWindow(cfg, &algos.InMemoryFixedWindowStore{Backend: b.mem})
}
func (b *memoryBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newRollingWindow(cfg, &algos.InMemoryRollingWindowStore{Backend: b.mem})
}
func (b *redisBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) {
return newTokenBucket(cfg, &algos.RedisTokenBucketStore{Backend: b.redis})
}
func (b *redisBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newFixedWindow(cfg, &algos.RedisFixedWindowStore{Store: b.redis})
}
func (b *redisBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newRollingWindow(cfg, &algos.RedisRollingWindowStore{Backend: b.redis})
}sourcecraft.dev
Далее остаётся написать функцию, создающую непосредственно саму middleware, а именно функцию возвращающую тип, поддерживающий serveHTTP метод:
package middleware
import (
"log"
"net/http"
"qwrttqr-rate-limiter/core/server/algos"
"qwrttqr-rate-limiter/core/server/facade"
"strconv"
)
func NewHTTPMiddleware(
algo algos.Limiter,
keyDer HTTPKeyDerivationFunction,
costDer HTTPCostDerivationFunction,
failOpen bool,
) func(http.Handler) http.Handler {
if algo == nil || keyDer == nil {
panic("middleware: limiter and key function are required")
}
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fail := func(stage string, err error) {
log.Printf("limiter error on %s: %v", stage, err)
if failOpen {
next.ServeHTTP(w, r)
return
}
http.Error(w, "service unavailable", http.StatusServiceUnavailable)
}
tokens := 1.0
if costDer != nil {
t, err := costDer(r)
if err != nil {
fail("cost derivation", err)
return
}
tokens = t
}
key, err := keyDer(r)
if err != nil {
fail("key derivation", err)
return
}
req, err := facade.NewRequest(key, tokens)
if err != nil {
fail("request creation", err)
return
}
decision, err := algo.Allow(r.Context(), req)
if err != nil {
fail("allow", err)
return
}
if !decision.Allowed {
w.Header().Set("Retry-After", strconv.FormatInt(decision.RetryAfter, 10))
http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
return
}
next.ServeHTTP(w, r)
})
}
}sourcecraft.dev
С gRPC необходимо реализовать 2 вида interceptor'а - унарный и потоковый:
func NewGRPCUnaryInterceptor(
algo algos.Limiter,
keyDer GRPCKeyDerivationFunction,
costDer GRPCCostDerivationFunction,
failOpen bool,
) grpc.UnaryServerInterceptor {
check := newGRPCCheck(algo, keyDer, costDer, failOpen)
return func(
ctx context.Context,
req any,
info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler,
) (any, error) {
if err := check(ctx); err != nil {
return nil, err
}
return handler(ctx, req)
}
}
func NewGRPCStreamInterceptor(
algo algos.Limiter,
keyDer GRPCKeyDerivationFunction,
costDer GRPCCostDerivationFunction,
failOpen bool,
) grpc.StreamServerInterceptor {
check := newGRPCCheck(algo, keyDer, costDer, failOpen)
return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
if err := check(ss.Context()); err != nil {
return err
}
return handler(srv, ss)
}
}sourcecraft.dev
И check:
func newGRPCCheck(
algo algos.Limiter,
keyDer GRPCKeyDerivationFunction,
costDer GRPCCostDerivationFunction,
failOpen bool,
) func(ctx context.Context) error {
if algo == nil || keyDer == nil {
panic("middleware: limiter and key function are required")
}
return func(ctx context.Context) error {
fail := func(stage string, err error) error {
log.Printf("limiter error on %s: %v", stage, err)
if failOpen {
return nil
}
return status.Error(codes.Unavailable, "service unavailable")
}
tokens := 1.0
if costDer != nil {
t, err := costDer(ctx)
if err != nil {
return fail("cost derivation", err)
}
tokens = t
}
key, err := keyDer(ctx)
if err != nil {
return fail("key derivation", err)
}
req, err := facade.NewRequest(key, tokens)
if err != nil {
return fail("request creation", err)
}
decision, err := algo.Allow(ctx, req)
if err != nil {
return fail("allow", err)
}
if !decision.Allowed {
md := metadata.Pairs("retry-after", strconv.FormatInt(decision.RetryAfter, 10))
_ = grpc.SetTrailer(ctx, md)
return status.Error(codes.ResourceExhausted, "rate limit exceeded")
}
return nil
}
}sourcecraft.dev
Как вы могли заметить, везде можно передать функции извлечение ключа и стоимости запроса. Это нужно так как в режиме middleware вы чаще всего будете писать сервисы, которые смотрят наружу, а значит будут принимать запросы от недоверенных клиентов.
Итог
Статья получилась достаточно объёмной, надеюсь она будет полезна ребятам, которые только начинают учить Golang)
исходник
Mentions
Wait a minute... There IS a rust implementation! apache/iceberg-rust
Why on earth is Apache Iceberg written in Java. What the hell is wrong with people. Need a Rust fork lmao apache/iceberg
Flink Agents 0.4.0で見るストリーミングエージェント この記事は Namsang Labs 技術ブログ からの転載です。 10月30日にApache Flink Agents 0.4.0がリリースされた。本番向けの1.0を控え、機能とAPIを整理するプレビュー版だ。今回の0.4.0には、私もコントリビューターとして参加した。リリースを機に、Flink Agentsがどんなプロジェクトで、エージェントをどう実行し、障害からどう復旧させるのかをまとめた。 Flinkの上でエージェントを実行する理由 決済が失敗したり、センサーが異常な値を送ってきたりしたら、システムはそのイベントを受け取り、原因を判断して必要な処置を進めなければならない。人がチャット画面で質問するまで待てる仕事ではない。Flink Agentsは、こうした処理のために始まったプロジェクトだ。流れ続けるイベントをエージェントが受け取って判断し、ツールを呼び出して必要な作業を行うことを目指している。 イベントが増えれば複数のサーバーに処理を分散する必要があり、顧客や機器ごとに、それまでに処理した内容も覚えておかなければならない。動き続けるシステムなので、サーバーに障害が起きたり実行が中断したりしても、作業を続けられる必要がある。モデルを呼ぶコードだけでは、この実行環境は作れない。 Flinkには、データストリームを分散処理し、キーごとに状態を管理しながら、障害時にはチェックポイントから復旧する機能がある。Flink Agentsはこのランタイムの上に、モデルとツール、メモリを使うエージェントを実行する仕組みを加えたApache Flinkのサブプロジェクトだ。イベントの受信からエージェントの判断、結果の処理までを、一つのFlinkジョブの中でつなぐ。 0.1.0から0.4.0まで 最初のプレビュー版となる0.1.0は、2025年10月にリリースされた。作業をアクションという単位に分けてイベントでつなぎ、Flinkの中でモデルやツールを呼ぶ基本の仕組みができた。実行を追うためのイベントログと、完了したアクションの記録を復旧時に再利用する機能も、この時点で含まれていた。Flink 1.20.3を基盤にPythonとJavaのAPIを提供していたが、初期の機能は主にPythonで実装されていた。 2026年2月の0.2.0ではJava APIの機能が補われ、対応するFlinkのバージョンも1.20から2.2まで広がった。別の言語で実装されたリソースも使えるようになった。メモリは、1回の実行だけで使うものと次の実行にも残すものを分けられるようになった。復旧時には、完了したアクションの実行記録に加えて、まだ終わっていないアクションの中で保存済みの個々の呼び出し結果も再利用できるようになった。 0.3.0は2026年6月にリリースされた。モデルやツールの設定をYAMLで宣言し、別の言語で書いたアクションも一緒に実行できるようになった。長期メモリはMem0を使って情報を抽出、整理する方式に変わり、作業の指示や参考資料を提供するAgent Skillsも追加された。実行記録の保存先としてApache Flussを選べるようになったほか、Web UIでイベントログを確認する機能も加わった。復旧時に結果の分からない外部呼び出しを確認する調整関数も、0.3.0で追加された。それまでは保存済みの呼び出し結果を再利用する方式だったが、呼び出しの途中で中断して結果が残っていない場合にも、外部システムの状態を確認する手段ができた。 今回リリースされた0.4.0では、そこにサブエージェントやモデルルーティングなどの機能が加わった。モデルを呼んだ後のロジックを同じアクションに書けるctx.chatも追加され、イベントだけでなく、モデルやツールの呼び出しまで追えるようになった。1.0からAPIの互換性を約束するのに先立ち、既存のAPIとデフォルトの動作も整理された。 エージェントを実行する仕組み エージェントはFlinkジョブの中のオペレーターとして実行される。前段からデータを受け取って作業し、結果を次のオペレーターに渡す。エージェントの内部でも、処理はイベントでつながっている。基本の構成要素はイベントとアクション、リソースだ。イベントが届くと条件に合うアクションが実行され、必要な処理をした後で新たなイベントを送る。一つのイベントに複数のアクションが反応することもあり、分岐や繰り返しを作れる。モデルやツール、プロンプトのように複数のステップで使うものは、リソースとして登録して名前で参照する。MCPサーバーのツールやプロンプトも接続でき、0.3.0からはリソースをYAMLで宣言できる。 商品レビューを要約するエージェントなら、入力レコードがInputEventとして届き、それを受け取ったアクションがChatRequestEventで要約を依頼する。組み込みのチャットアクションがモデルを呼んでChatResponseEventを送ると、応答を処理するアクションが要約文をOutputEventに入れる。その値がFlinkの次のオペレーターに渡る。こうしてステップ同士をイベントでつなぐので、イベントの記録をたどれば、どの入力からどんな判断と出力が続いたのかを確認できる。 処理の順序をコードで決めることも、次の行動をモデルに任せることもできる。組み込みのReActAgentにはモデルを指定し、必要に応じてツールやプロンプトを加える。モデルは目標を達成するために必要なツールを選び、その結果を見て次の行動を決める。全体の手順はコードで決め、一部の調査だけをこうしたエージェントに任せる形で、二つの方式を組み合わせることもできる。 エージェントを定義したら、Flinkのジョブにつなぐ。0.4.0のドキュメントのPython例から、その部分を抜き出すと次のようになる。 env = StreamExecutionEnvironment.get_execution_environment() agents_env = AgentsExecutionEnvironment.get_execution_environment(env) output_stream = ( agents_env.from_datastream( input=input_stream, key_selector=lambda x: x.id ) .apply(your_agent) .to_datastream() ) 入力はキーごとに分かれ、前段から届いたレコードごとに新しい実行が始まる。ドキュメントではこれをagent runと呼ぶ。上のコードのx.idが顧客IDなら、同じ顧客の入力同士でメモリを共有するように構成できる。キーを選ぶ関数を渡せばその関数を使い、省略すればDataStreamにすでに指定されているキーを使う。どちらもなければ接続できない。Tableも接続できるが、入力テーブルは行の追加だけを行うものに限られる。既存の行を更新したり削除したりする集計結果は、そのままではつなげられない。 メモリは、どれだけ長く残し、どう読み出すかによって分かれる。感覚メモリと短期メモリは、どちらもFlinkの状態に保存される。感覚メモリは1回の実行が終わると消え、短期メモリは同じキーで届いた次の実行でも読み出せる。顧客ごとのエージェントなら、今回のリクエストを処理するあいだだけ必要な値は感覚メモリに、次のリクエストでも使う値は短期メモリに置く。 長期メモリは複数の実行で得た情報を整理して残し、セマンティック検索で探す。0.3.0からはMem0が情報の抽出と整理を担当し、実際のデータは外部のベクトルストアに保存する。そのため、チャットモデルと埋め込みモデル、ベクトルストアをあわせて設定する必要がある。長期メモリの検索範囲も、デフォルトでは同じジョブと入力キーの中に限られる。 モデルに作業の指示を与えるAgent Skillsもある。SKILL.mdに書いた指示と関連するスクリプト、参考資料をまとめておくと、エージェントが必要に応じて内容を読み込んで使う。 障害から復旧する仕組み エージェントは一つの作業を終えるまでに、モデルや外部ツールを何度も呼ぶことがある。その途中でタイムアウトや呼び出し回数の制限にかかることもあれば、エージェントを実行していたプロセスが止まることもある。呼び出しエラーへのリトライ方針とは別に、実行が中断した後、完了済みの作業を繰り返さずに残りを続ける方法も必要になる。最初から実行し直しても、前と同じ結果になるとは限らない。同じ入力でもモデルが別の答えを返すことがあり、外部システムから取得した値も、そのあいだに変わっているかもしれない。送信済みのメールや決済リクエストまで繰り返せば、外部の作業が重複する可能性もある。復旧するには、何が完了していて、何を再実行する必要があるのかを区別しなければならない。 Flinkは障害が起きると、最後のチェックポイントから状態と入力の位置を復元し、それ以降の入力を再処理する。外部ストアに結果を書き込むときは、トランザクションや冪等性に対応したシンクを使い、同じ結果が重複して反映されるのを防げる。だがエージェントが入力を処理し直すと、モデルやツールも再び呼ぶことがあり、すでに行った外部の作業をチェックポイントで元に戻すことはできない。 完了したアクションの実行記録を復旧に使うには、Flinkのチェックポイントに加えて、アクション状態ストア(Action State Store)を設定する。KafkaやApache Flussにアクションの実行記録を残す仕組みで、完了したアクションの記録を再利用する機能は0.1.0から提供されている。0.4.0の一貫性に関するドキュメントによれば、完了記録のあるアクションは本体を実行し直さず、保存しておいた感覚メモリと短期メモリの変更、出力イベントをそのまま再生する。ただし、非同期で実行されていた複数のアクションが共有のメモリを読み書きした順序までは再現しない。その順序で結果が変わるなら、必要な順序をイベントの依存関係として明示しなければならない。 0.2.0では、復旧時に再実行しない範囲をアクションの中の呼び出し単位まで狭めた。アクションが終わっていなくても、その中で耐久実行(durable execution)で包んだ呼び出しが終わって結果が保存されていれば、復旧時にその呼び出しはもう行わない。アクションのドキュメントはプロフィールの取得を例に挙げている。 class ProfileAgent(Agent): @action(EventType.InputEvent) @staticmethod def process_input(event: Event, ctx: RunnerContext) -> None: user_id = InputEvent.from_event(event).input profile = ctx.durable_execute( profile_service.get_profile, user_id, durable_id=f"profile:{user_id}", ) ctx.send_event(OutputEvent(output=profile)) プロフィールの取得が終わり、結果まで保存された後でアクションが失敗すると、復旧したアクションはサービスを呼び直さず、保存されたプロフィールを取り出して使う。ドキュメントにもあるとおり、耐久実行の呼び出しの外にあるコードは再実行されることがある。完了記録と呼び出しの結果は外部ストアに同期的に書き込まれ、復元するチェックポイントと、その復旧に必要な実行記録も一緒に保持しておかなければならない。復旧後の実行が以前の記録を見つけて使うには、キーごとの入力順序とキーの型、シリアライザの設定、アクションの名前が変わっていないことが条件になる。耐久実行の呼び出しも同じ入力と順序、安定した識別子を保つ必要があり、保存された値は復元の時点でも読める状態でなければならない。 呼び出しを始めてから結果を保存するまでのあいだに障害が起きると、呼び出しが実際に成功したかどうかは記録だけでは分からない。外部サービスが同じリクエストを繰り返しても処理を重複させないなら、再試行できる。そうでなければ、そのリクエストが処理されたかを先に確かめる必要がある。0.3のロードマップ解説もこの問題を取り上げ、ユーザーが外部の状態を確認するための関数を提案していた。 0.3.0では、この確認処理を実装できる調整関数(reconciler)が追加された。復旧時に、実行中の呼び出しの記録はあるのに結果がなければ、この関数が実行される。調整関数には、外部システムの処理状況を確かめ、元の結果を返すか失敗を知らせるコードを書く。こうして得た結果も記録に残る。調整関数がなければ、元の呼び出しをもう一度実行する。 アクション状態ストアは、外部の動作とその結果を保存する処理を一つのトランザクションにまとめられない。そのため外部の状態を変える呼び出しは、繰り返されうるものとして設計する必要がある。外部サービスの冪等性の仕組みを使うか、耐久実行の呼び出しに元の結果を突き止められる調整関数を付けるのが手だ。 新たに追加された機能 今回のバージョンでは、作業を別のエージェントに任せたり、リクエストに合うモデルを選んだりする機能が追加された。モデル呼び出しと応答処理を一つのアクションに書けるようになり、運用中にモデルやツールの呼び出し記録も確認できるようになった。 表のテキスト版は元の記事で読めます。 0.3.0のリリース後にあたる2026年9月15日、TypeSafeはJevを公開した。与えられた文脈をあらかじめ決めた基準に照らして判断し、確率や選択肢、スコアとして結果を返すモデルだ。Flink Agentsは今回のバージョンに意思決定モデルAPIを追加し、JevとOpenAI、アリババクラウドのModel Studioを同じインターフェースで使えるようにした。たとえば、顧客の問い合わせを請求担当と技術サポートのどちらに回すか判断したり、あらかじめ決めた尺度でアラートのリスクを評価したりできる。モデルに渡す事実と判断基準、結果を受けて実行する後続の処理は、アプリケーション側で決める。 開発ツールとモデル連携も補強された。Javaの非同期実行は、JDK 21に加えてJDK 11と17にも対応した。チャットモデルは、JavaでGeminiを、JavaとPythonでvLLMとIBM watsonx.aiを新たに使える。JavaにはOpenAIの埋め込みモデルも追加された。コーディングエージェントがFlink Agentsのアプリケーションを作って検証するときに参照するflink-agents-devスキルも追加された。 これらの機能を使って、深刻なアラートだけを選んで処理するエージェントを作るとしよう。条件トリガーで、深刻度の高いアラートに対してだけアクションを実行する。そのアクションの中でctx.chatを使ってモデルを呼び、応答をその場で処理する。モデルが登録済みのツールを使いたいと返せば、フレームワークが代わりに実行する。 イベント駆動のチャットAPIでは、リクエストと応答をそれぞれ別のアクションで処理する。この方式は0.4.0でも引き続き使える。ctx.chatは、モデルを呼んだ後のロジックを同じアクションに続けて書きたいときに使うAPIだ。耐久実行を有効にしておけば、ctx.chatの応答も復旧時に再利用できる。 追加の調査はサブエージェントに任せられ、その呼び出しも耐久実行に対応している。決まった基準でリスクを評価したいなら意思決定モデルを使い、リクエストごとに別のモデルを使いたいならモデルルーティングを使う。単純なリクエストは軽いモデルに送り、複雑な作業にだけ大きなモデルを使う、といった使い分けだ。 運用では、トレースログで一つの入力がどのステップや呼び出しを経て処理されたのかを確認する。組み込みメトリクスで失敗や遅延を見つけたら、このログでどのステップに問題があったのかを絞り込める。デフォルトの記録範囲はイベントだけなので、ステップや呼び出しまで見るにはtrace-log.targetsを設定する必要がある。イベント同士の因果関係をツリーの形に描き直すツールも追加された。このツールはイベントの記録だけを読み、実行ステップや呼び出しの記録は読み飛ばす。トレースログの書き込みに失敗しても、イベント処理とジョブの実行は続く。だから、トレースログだけを見てすべての実行が漏れなく記録されたと言い切ることはできない。 0.4.0へのアップグレード方法 0.xはまだプレビュー版なので、APIや設定が互換性のない形で変わることがある。今回は、初期の本番ユーザーからのフィードバックやAPIレビューで分かった問題を整理したため、互換性のない変更がこれまでのどのリリースよりも多くなった。0.4.0を安定させてから小さな調整を経て1.0をリリースし、そこからAPIの互換性を約束する計画だ。具体的な互換性ポリシーは別途発表される予定になっている。0.3.xから0.4.0にアップグレードするときは、まず次の変更点を確認する必要がある。 アクションの宣言: Javaの@Action(listenEventTypes = ...)は@Action(...)に、Pythonのadd_action(events=...)とYAMLのlisten_toはtrigger_conditionsに変わる。よく使うPythonの@action("event_type")という宣言はそのまま使える。 Pythonの実行経路: from_list()とto_list()でFlinkなしに実行していた経路がなくなった。FlinkのStreamExecutionEnvironmentにDataStreamかTableでつなぐ必要があり、ローカルで開発するときはローカルのFlinkランタイムを使う。 リトライのデフォルト: フレームワークのmax-retriesのデフォルトが3から0になり、error-handling-strategyはなくなった。失敗したチャットの応答はアプリケーション側で処理しなければならない。モデル提供者のSDKのリトライは別に設定する。 並行実行のデフォルト: デフォルトの非同期設定では、互いに独立したツール呼び出しが並行して実行されることがある。順序が大事なツールなら、ステップを分けるか並列度を絞る必要がある。JDK 11から20でJavaのアクションだけを使う場合は、実験的な機能であるアクションの並列実行がデフォルトで有効になり、parallel-execution.enabledで無効にできる。 ツール引数の検証: 未知のフィールドや誤った型、不正な値が来ると実行前に失敗する。数値の文字列を数値に変換することもない。 ログの設定: イベントログがトレースログに置き換わり、event-log.のような以前の設定キーは受け付けられない。trace-log.のキーに移す必要がある。 実行記録: 0.3.xが残したアクション状態ストアのデータは0.4.0と互換性がないので、以前の実行記録を再利用しないアップグレード計画が要る。Flinkのチェックポイントやセーブポイントを復元できるかどうかは、このストアのデータ互換性とは別に確認する必要がある。 処理中のアクション: 処理中のアクションを含むオペレーターのスナップショットは、0.4.0にアップグレードするときに復元に失敗することがある。以前のバージョンにもあった制約なので、実際に近い状態でまず復元を試しておくのがよい。 メッセージの形式も変わった。ChatMessageは内容をblocksに、追加の属性をmetadataに持つ。モデル呼び出しは、応答メッセージとモデルの情報、使用量を分けて持つChatResultを返し、ツールの結果はToolResponseに統一された。モデル連携を自作していたり、メッセージをJSONで保存していたりするなら、まずこの部分から手を入れる必要がある。 コントリビューターから見た0.4.0 告知のコントリビューター一覧には38人が載っており、私も今回のリリースから貢献できた。今回取り込まれた私のPR 23件のうち5件は、先に説明した耐久実行の再生経路を直したものだった。記録された結果がnullだと再生時に例外が出ていた #1171。pickleで保存はできるが復元できないPythonの例外は、見当違いのTypeErrorとして再生されていた #1182。調整関数なしで完了結果だけを記録するPythonの耐久実行の呼び出しは、Java側で結果の保存に失敗しても呼び出し元に知らせていなかった #1211。 通常の呼び出しと耐久実行の呼び出しを非同期でまとめて待っているときに、通常の呼び出しがキャンセルされると、入力順でそれより前にあった耐久実行の呼び出しは完了していても結果が保存されないケースがあった #1239。組み込みのPythonチャット呼び出しの識別子が、プロセスごとに変わりうる問題も直した #1240。そのほか、モデルルーティングの検証やメモリの参照、ランタイムのキュー、メトリクス、ドキュメントを修正した。 リリースの直前には最初のリリース候補を検証し、投票にも参加した。その経緯は「オープンソースにも投票がある」に別に書いた。検証中に、バージョンを上げた後で既存のスナップショットを復元するときの問題を確認し、その内容が正式な告知のアップグレード時の注意事項に反映された。その後、バージョンをまたぐ状態の復元を検証するテスト基盤を作る作業も担当することになった。1.0ではAPIが安定し、アップグレードの検証も充実して、運用中のエージェントを今より保守しやすくなることを期待している。 始めるには 実行環境の設定で使用するFlinkとJava、Pythonのバージョンを確認したうえで、最初のエージェントを作るの例を実行すればよい。処理の手順を書くより先にモデルにツールを任せてみたいなら、ReActAgentのクイックスタートが向いている。コーディングエージェントで開発するなら、flink-agents-devスキルを追加できる。これはコーディングエージェントがAPIやサンプルを参照するための開発資料であり、FlinkとFlink Agentsの実行環境は上の案内に従って別にインストールする。 npx skills add 'apache/flink-agents#release-0.4@flink-agents-dev' バグや機能の要望はGitHubのイシューで受け付けており、利用事例やアイデアはGitHubのディスカッションと、Apache FlinkのSlackにある#flink-agents-userチャンネルで話せる。 参考資料 Apache Flink Agents 0.4.0 Release Announcement、Wenjin Xie、2026年。0.4.0で追加された機能と互換性のない変更、そして0.4.0が1.0前の最後のリリースだという計画を書いている。 Apache Flink Agents 0.3.0 Release Announcement、2026年。Flinkの上でエージェントを実行する理由と、0.4.0と1.0へと続くAPI互換性の計画を説明している。 Apache Flink Agents 0.2.0 Release Announcement、2026年。3種類のメモリと呼び出し単位の耐久実行、複数のFlinkバージョンへの対応を紹介している。 Apache Flink Agents 0.1.0 Release Announcement、2025年。プロジェクトがなぜ必要なのかと、最初のプレビュー版の主な機能を紹介している。 Flink Agents 0.3 Roadmap解读、Flink中国語コミュニティ、2026年。0.3.0のロードマップをコントリビューターの視点で解説し、障害復旧で残る問題と解決策の候補を挙げている。同コミュニティのこの記事には、まだ0.4.0の内容が反映されていない。 Flink Agents 0.4.0のドキュメント。はじめにと開発ガイド、一貫性と復旧の流れを説明する運用の章がある。 広告やCRMからAIエージェントまで、大規模なプラットフォームを作り、率いてきた経験を Namsang Labs 技術ブログ に書いています。
Lucene 10.6.0 の doc-values updates は列ごとの書き直しをどうやめたのか はじめに OpenSearch や Elasticsearch では、検索したいデータを JSON のドキュメント（doc）として index に保存します。 doc は title や price のような field の集まりで、OpenSearch や Elasticsearch はこの field の値を使って検索、ソート、集計をします。 この index を実際にディスクに書き、検索するのが、Java で書かれた検索ライブラリの Apache Lucene です。 Lucene は index を segment という単位のファイル群に分けて書き、一度書いた segment は変更しません。 こうして保存する doc には、一部の field だけが頻繁に変わるものがよくあります。 EC の在庫数や価格、記事の閲覧数、ランキング用の人気スコアなどです。 doc 本体には本文や 512 次元の vector が入っていて、変わるのは 8 バイトの数値 1 つだけ、という形です。 新しい doc は新しい segment に書かれ、既存の segment のファイルはそのまま残ります。 Lucene でこの数値を変える方法は 2 つあります。 1 つは IndexWriter#updateDocument で doc ごと入れ直す方法で、本文も vector も index し直すことになります。 もう 1 つは IndexWriter#updateNumericDocValue で、doc-values の値だけを書き換える方法です。 doc-values は、Lucene が field の値を doc ID 順に並べて列として持つ構造です。 全文検索に使う転置インデックスは「語から doc」を引く向きの構造で、doc-values はその逆の「doc から値」を引く向きです。 ソート、ファセット、集計は、この列を doc ID で引いて値を読みます。 OpenSearch や Elasticsearch の mapping にある doc_values は、この列を作るかどうかの設定で、text 以外の多くの field では default で有効です。 updateNumericDocValue は doc を入れ直さないので、軽い操作に見えます。 ところが Lucene 10.5 までの doc-values update は、別のものを丸ごと書き直していました。 更新した doc の数に関係なく、その field の列（column）を segment 1 つぶん書き直すのです。 2026 年 10 月 29 日にリリースされた Apache Lucene 10.6.0 で、この書き直しをやめる仕組みが入りました。 CHANGES.txt の New Features には次のように書かれています。 GITHUB#16418: Incremental doc-values updates. A set-only doc-values update (NUMERIC or BINARY) is written as a sparse "delta" overlay holding just the updated documents and layered over the base column at read time, instead of rewriting the whole column, turning per-update write amplification from O(column) into O(updated docs). Opt-in; enable with IndexWriterConfig#setMaxDocValuesOverlays(n). 気になるのは、書き込み量がどれだけ減り、代わりに何を払うのか、そして OpenSearch と Elasticsearch にどう関わるのかです。 元になった Pull Request #16418 と、10.5.0 と 10.6.0 のソースを読んで確かめました。 ! 本文のクラス名とメソッド名は Lucene 10.5.0 と 10.6.0 のタグのものです。 issue や PR の状態は 2026 年 11 月 1 日時点で確認しました。 setMaxDocValuesOverlays には @lucene.experimental が付いており、今後挙動が変わる可能性があります。 1 doc を更新すると何が書かれるのか gen ファイルに書かれる列 doc-values の列も segment の一部なので、segment と同じく一度書いたら変更しません。 値を更新するには、別のファイルを書き足す必要があります。 Lucene はこの書き足しを generation（gen）という番号で管理しています。 doc-values を更新すると、その field について新しい gen のファイルを書き、読み取り時には最新の gen を使う仕組みです。 10.5.0 の ReadersAndUpdates#handleDVUpdates では、次の gen 番号を 36 進数の文字列にしてファイル名の suffix にしています。1 final long nextDocValuesGen = info.getNextDocValuesGen(); final String segmentSuffix = Long.toString(nextDocValuesGen, Character.MAX_RADIX); 問題は新しい gen ファイルの中身です。 同じメソッドの中で、MergedDocValues が既存の列の値と更新値を doc ID 順に混ぜ合わせ、その結果を新しい gen として書き出します。 更新されなかった doc の値も含めて、列全体です。 新しい gen ファイルの大きさは列全体（値を持つすべての doc）で決まり、更新した doc が 1 件でも 3 件でも同じです。 PR の説明はこの状態を「Lucene rewrites the whole column for that field, even if you only touched a few docs」と書いています。 segment が大きくなるほど、1 回の更新で書く量も増えます。 soft delete の場合 この書き直しは明示的に doc-values を更新したときだけの話ではありません。 Lucene の soft delete は、削除の印を soft-deletes 用の numeric doc-values field に書くことで実現しています。 つまり soft delete も doc-values update であり、同じ経路で列を書き直します。 soft-deletes の列は印を付けた doc だけを持つので、1 回に書く量は数値 field の列ほど大きくありません。 Elasticsearch は操作の履歴を soft delete した doc として Lucene に残しています。 shard の復旧や CCR（cross-cluster replication）で、この履歴を使うためです。 後で見る Elasticsearch の PR #157749 の説明にも、この仕組みが出てきます。 OpenSearch は、この仕組みを持つ Elasticsearch 7.10 から分岐したプロジェクトです。 そのため、doc-values を明示的に更新していない index でも、update や delete で doc を置き換えるたびにこの経路を通ります。 10.6.0 で入った overlay 読み取り時の重ね方 PR の説明によると、10.6.0 で入った仕組みの考え方は単純です。 値を設定するだけの更新（set-only）なら、変わった doc だけを持つ疎な（sparse）差分を新しい gen として書きます。 PR ではこの差分を delta、読み取り時に重ねる層を overlay と呼んでいます。 先ほどの図と同じ更新でも、書くのは変わった doc の数だけになります。 読み取り時は base の列の上に delta を新しい順に重ね、同じ doc が複数の層にあれば最新の値を使う、という読み方です。 PR の説明はこれで「Updating a field becomes O(changed docs) instead of O(column)」になると書いています。 読み取り側は新しく入った OverlayNumericDocValues と OverlayBinaryDocValues（10.6.0 では OverlaySortedNumericDocValues も）が担当します。 delta は codec に元からある疎な形式で書くので、PR は「No doc-values format change」と書いています。 overlay が溜まったときの fold delta を書き足すだけだと、層がどんどん積み上がり、読み取りが遅くなります。 そこで field ごと、segment ごとに、次の決まりで delta を畳みます（fold）。 flush のたびに、変わった doc だけの delta を 1 つ書く delta の数が上限に達していたら、次の更新を書くときに既存の delta と今回の更新を 1 つにまとめる fold のとき、既存の delta が持つ doc 数の合計が segment の doc 数の 50% 以上なら、base と合わせて新しい密な（dense）base を書く segment の merge で、普通の列に戻る 上限の数は後で見る setMaxDocValuesOverlays で指定します。 最初の実装では、dense な base に戻すのは delta が列全体を覆ったときだけでした。 review で neoremind が bug を見つけます。 一度 dense な base に fold すると、それ以降の fold がすべて列全体の書き直しに戻っていました。 作者は判定を segment に対する割合に変え、閾値を 50% にして直しています。 overlay にならない更新 すべての更新が delta になるわけではありません。 PR #16418 の時点の対象は NUMERIC と BINARY の doc-values でした。 10.6.0 では #16581 で SORTED_NUMERIC も単一の値で更新できるようになりました。 この更新は、列が単一値か複数値かに関係なく同じ delta の経路を通ります（複数値の doc は読み取り時に doc 単位で保たれます）。2 OpenSearch や Elasticsearch の数値 field は SORTED_NUMERIC の列として書かれるので、この拡張も関係します。 どの型でも対象は値を設定するだけの更新に限られ、値を消す（unset する）更新が混ざると、従来どおり列全体を書き直します。 ベンチマークで見る効果と代償 更新の間隔による書き込み量の違い PR #16418 には、作者自身のベンチマークが載っています。 対象は 8 バイトの数値と 512 次元の vector を持つ 5M（500 万）doc です。 4 スレッドがランダムな順序で更新し、NRT reader を 1 秒ごとに refresh します。 比べている方法は 3 つです。 full reindex は updateDocument で doc ごと入れ直す方法、dense は 10.5 までの doc-values update、sparse は今回の overlay です。 作者は、書き込み量が更新の間隔で大きく変わると考え、全速で更新する batch と、毎秒 5,000 件に絞った throttled の 2 通りで測っています。 数値 field の結果です。3 条件 方法 bytes/update 書き込みの倍率 batch full reindex 13,138 1,642× batch dense 80 10× batch sparse 8 1.0× throttled 5k/s full reindex 14,359 1,795× throttled 5k/s dense 7,202 900× throttled 5k/s sparse 18 2.2× 倍率は書いたバイト数を実際に変わった 8 バイトで割った値です。 batch では dense でも 10× で済んでいます。 refresh 1 回のあいだに大量の doc が更新されるので、列全体の書き直し 1 回を多くの doc で分け合えるからです。 throttled にすると、同じ書き直しが数千件ぶんしかカバーしなくなり、900× に跳ね上がります。 sparse は間隔に関係なく変わった doc だけを書くので、どちらでも 1.0〜2.2× に収まります。 throttled の dense と sparse を比べると、書き込み量は約 400 分の 1 です。 throttled の行では、dense と full reindex の差も小さくなっています。 doc ごと入れ直す 14,359 バイトに対して、doc-values update は 7,202 バイトで、半分にしかなっていません。 32 バイトの binary field では、さらに逆転します。 条件 方法 bytes/update throttled 5k/s full reindex 15,756 throttled 5k/s dense 27,734 throttled 5k/s sparse 55 このベンチマークの binary field では、更新が少しずつ流れ続けると、10.5 までの doc-values update が doc ごと入れ直すより多く書いていたことになります。 作者は throttled を「closer to a steady trickle on a live index」と位置づけていて、運用中の index に近いのはこちらの条件です。 soft delete でも同じ傾向が出ていて、throttled では dense の 33 bytes/update が sparse で 6 bytes/update になっています。 読み取りの遅れ 書き込みが減るぶん、読み取りは遅くなります。 読み取りのたびに base と delta を重ね合わせるからです。 同じベンチマークで、500 万 doc の field を頭から読む scan の時間はこう変わりました。 field 条件 dense sparse 数値 batch 35 ms 118 ms 数値 throttled 5k/s 21 ms 103 ms binary 32B batch 29 ms 327 ms binary 32B throttled 5k/s 28 ms 172 ms scan は 3〜11 倍遅くなっています。 作者はソートやファセットのように doc を飛び飛びに引く読み方も測っています。 advanceExact で 10% の doc を引く測定では、overlay が 16 層のとき約 18 ms で、通常の列は約 1 ms でした。4 throttled で約 350 の gen が残った状態では、約 26 ms 対約 7 ms です。 作者はこれを「tens of ns per lookup」と書き、クエリ全体の時間と比べれば小さいとしています。 ただし、どちらの数値も作者が PR の review のために書いたベンチマークのものです。 benchmark のコードは merge 前に外すと書かれていましたが、2026 年 11 月 2 日時点では main のテストソースに、単独で実行する IncrementalDocValuesUpdatesBenchmark として残っています。 luceneutil や nightly benchmark のような第三者の測定は見つかりませんでした。 自分の index で書き込みの減少が読み取りの遅れに見合うかは、更新の間隔と読み取りの重さを比べて決めることになります。 overlay を有効にする設定 overlay は IndexWriterConfig#setMaxDocValuesOverlays(int) で有効にします。 0 は無効で、従来どおり列全体を書き直す設定です。 1 以上の値を渡すと有効になり、その数まで overlay を溜めてから fold します。 10.6.0 の Javadoc はこの値のトレードオフを次のように書いています。5 higher values lower write amplification but keep more overlays (hence more files) live to merge at read time. 10.6.0 の default は 0 で、この機能は opt-in です。 setMaxDocValuesOverlays の Javadoc には「Only takes effect when IndexWriter is first created」ともあり、6動いている IndexWriter に後から設定しても効きません。 ! 10.6.0 のリリースノートの Highlights には「This is on by default」と書かれていますが、10.6.0 のソースの default は 0 です。 backport の PR #16554 も「opt-in on 10.x」と明記しています。 on by default なのは 11.0 系（branch_11x）と main（12.0）で、default は #16648 で 16 から 8 に下がっています。 アップグレード後に古い Lucene で開けるか overlay を有効にしなくても、10.6.0 で変わることが 1 つあります。 segments ファイルの format version が上がりました。7 /** The version that records per-field incremental doc-values overlay generations. */ public static final int VERSION_10_6 = 11; static final int VERSION_CURRENT = VERSION_10_6; 古い Lucene は overlay を読めないので、delta を列全体と取り違えないよう、format version で弾く必要があります。 PR #16554 はこの version を機能を使うかどうかに関係なくすべての commit で書くと説明しています。 So any index written by 10.6+ is rejected by older 10.x readers with IndexFormatTooNewException, whether or not it uses the feature. Lucene は新しいバージョンで書いた index を古いバージョンで読めることを元から約束していません。 そのため規約の上では何も失われていない、というのが PR の立場です。 運用側から見ると、10.6.0 で一度でも commit した index は、10.5 系の Lucene に戻した時点で開けなくなります。 ローリングアップグレードの途中で旧版に戻す手順を用意しているなら、10.6.0 からは使えないと考えたほうが安全です。 review では ChrisHegarty が、commit 時に dense へ fold すれば古いバージョンでも読めるのでは、と提案しています。 作者は、refresh と commit を同じ頻度で行う使い方ではcommit 時に fold する案ではこの機能の効果が消えるとして、採用しませんでした。 同じリリースの .dvp 分離 10.6.0 では doc-values のファイル構成も別の変更で変わっています。 PR #16540（issue #16298）は、疎な field の doc ID 集合（IndexedDISI）を .dvd から新しい .dvp ファイルに分けました。 overlay とは独立した変更ですが、どちらも「doc-values まわりのファイルが変わった」と読めるので、混同しないよう気をつけてください。 Lucene90DocValuesConsumer や Lucene90DocValuesProducer のコンストラクタを直接呼んでいるコードは、こちらの変更に合わせて直す必要があります。 OpenSearch の star-tree がその例です。 OpenSearch と Elasticsearch の対応状況 OpenSearch OpenSearch は PR #23475 で Lucene を 10.5.2 から 10.6.0 に上げ、2026 年 10 月 31 日に main へ merge しました。 Version.V_3_10_0 が LUCENE_10_6_0 を指すようになっています。 PR で直したコードの 1 つは、前の節で触れた .dvp 分離に合わせた star-tree の doc-values の対応です。 OpenSearch は overlay の設定には手を入れていません。 PR の diff にも、GitHub のコード検索で見た OpenSearch のリポジトリにも setMaxDocValuesOverlays は出てきません。 Lucene 10.6.0 は default の 0、つまり overlay が無効のまま取り込まれています。 一方、segments format の変化は、機能の有無に関係なく OpenSearch にも及びます。 PR には書かれていませんが、Lucene 側の仕様から考えると、OpenSearch 3.10 で書いた index は Lucene 10.5 系の OpenSearch では開けなくなるはずです。 Elasticsearch Elasticsearch には、10.6.0 への upgrade PR は見つかりませんでした。 Elasticsearch は lucene_snapshot というブランチで Lucene の開発版に追従しているので、公開リポジトリ上の取り込み方が OpenSearch とは違います。 代わりに、overlay を使う機能の提案があります。 #16418 の作者である jimczi による PR #157749「Add updatable doc-values fields」で、2026 年 11 月 1 日時点で open のままです。 feature flag doc_values_updatable の内側にあり、ラベルは v9.7.0 です。 この PR は mapping に doc_values: { updatable: true } を足し、_update や _bulk の部分更新を doc の入れ直しではなく doc-values の書き換えで行います。 対象は keyword、数値、date の field で、index: false であること、index sort に含まれないこと、sub-field を持たないことが条件です。 PR が「columnar indices only」と書く種類の index で、新しい index version（IndexVersions.DOC_VALUES_UPDATABLE）以降に作ったものに限られます。 PR はねらいを「avoiding the cost of rewriting large immutable fields such as vectors」と書いていて、はじめにで挙げた「vector を持つ doc の数値 1 つだけを変える」場面そのものです。 overlay の上限は index 設定 index.doc_values.max_overlays で決めます。 doc-values の書き換えだけで済ませるには、Elasticsearch 側にも工夫が要ります。 _seq_no は in-place で書き換えられないので、更新の sequence number を _dv_update_seq_no という別の列に記録しています。 また、更新の差分だけを replica に送っても、replica が base の doc を持っていなければ意味がありません。 そのため、base の doc が global checkpoint 以下（すべての in-sync なコピーが持っていることが保証される範囲）にあるときだけ差分を送り、それ以外は従来の reindex に戻しています。 関連する変更と残っている課題 #16418 の後にも、いくつかの変更と議論が続いています。 ファイル数の問題は issue #16644 で dweiss が報告しました。 テストがファイルハンドルの上限に引っかかったというもので、issue に引用された分析は、開いているファイルの数が「segments × 更新した field 数 × (1 + overlay 数) × 2」に比例して増えるという見積もりを書いています（原文は英語）。 作者の jimczi は、増え方は field ごとに overlay の上限で抑えられると答えています。 issue はテスト側で overlay 数をランダムにする #16645 で閉じられ、これとは別に #16648 で default も 16 から 8 に下がりました。 overlay を有効にするなら、OS のファイルハンドル上限も合わせて確認しておくと安全です。 merge 中の更新の扱いも変わりました。 10.5 までは、merge 中に来た doc-values update を ReadersAndUpdates が heap に溜めていました。 #16570 はこれをやめ、merge の完了時に元の segment から読み直すようにしています。 10.x への backport（#16591）では、この挙動は overlay を有効にしたときだけ働きます。 10.6.0 には、SORTED_NUMERIC の更新のほかにも関連する変更が入っています。 #16565 は index sort に使う field への binary update を拒否するようにしました。 PR #16418 の review で後回しにされた項目と、10.6.0 のソースに TODO として残っている項目もあります。 reader を開くときに delta を 1 つにまとめる。shaie の提案に、作者は「it's the top follow-up」と答えています8 読み取りで intoBitSet を使う、dense な base を直接読んで変わった doc だけ差し替える9 field ごとに設定を変えられるようにする。ChrisHegarty の提案で、作者は「Good follow-up」と答えています10 毎回すべての delta を畳み直すのをやめ、size-tiered な fold にする11 fold の閾値 50% を設定で変えられるようにする12 確認できていないこと dev メーリングリスト13に、この機能を扱った専用のスレッドは見つかりませんでした。確認できたのはリリースの告知だけです。 ベンチマークの数値は作者のものだけで、第三者の測定は見つかっていません。 overlay を有効にしたときに、実際の index でファイルハンドルがどれだけ増えるかは測っていません。 Elasticsearch が Lucene 10.6.0 をどのバージョンで取り込むかは確認できていません。 おわりに はじめにの在庫数の例に戻ります。 10.5 までの updateNumericDocValue は在庫数を数件変えただけでも、refresh のたびに segment ぶんの在庫数の列を書き直していました。 10.6.0 で overlay を有効にすれば、書くのは変わった在庫数だけになります。 その代わり、在庫数でソートや集計をするたびに、base に delta を重ねる手間がかかります。 10.6.0 では opt-in なので、何もしなければこの挙動は変わりません。 ただし segments format は必ず上がるので、10.6.0 で一度 commit した index は 10.5 では開けなくなります。 Lucene を直接使っているなら、まず自分の index の更新の間隔と、ソートやファセットの重さを見比べてみてください。 更新が少しずつ続き、読み取りが軽いなら、setMaxDocValuesOverlays に小さな値を渡して試す価値があります。 参考リンク Release 10.6.0 · apache/lucene Lucene Core News CHANGES.txt（10.6.0） Incremental doc-values updates · PR #16418 Backport incremental doc-values updates to 10.x · PR #16554 Incremental doc values increase open file handle count significantly · Issue #16644 Upgrade Lucene to 10.6.0 · OpenSearch PR #23475 Add updatable doc-values fields · Elasticsearch PR #157749 脚注 ReadersAndUpdates.java#L323-L324（10.5.0） ↩︎ OverlaySortedNumericDocValues.java（10.6.0） の Javadoc。CHANGES.txt の GITHUB#16580 は「Single-valued columns reuse the numeric update path」と単一値の列に限った書き方をしていますが、コードは複数値の列にも delta を書きます。 ↩︎ PR #16418 の Numbers 節。この節の表と数値は、特記しない限りここから引いています。 ↩︎ PR #16418 の作者のコメント。「tens of ns per lookup」もこのコメントからの引用です。 ↩︎ IndexWriterConfig.java#L111-L116（10.6.0） ↩︎ IndexWriterConfig.java#L336-L348（10.6.0） ↩︎ SegmentInfos.java#L129-L132（10.6.0） ↩︎ 作者の回答 ↩︎ 作者のコメント ↩︎ ChrisHegarty の提案、作者の回答 ↩︎ ReadersAndUpdates.java#L390-L391（10.6.0） の TODO ↩︎ ReadersAndUpdates.java#L58-L60（10.6.0） の TODO ↩︎ Lucene の開発者向けメーリングリスト dev@lucene.apache.org です。Apache Software Foundation のプロジェクトでは、リリースの投票や設計の議論をこのリストで行います。アーカイブは lists.apache.org で検索できます。確認できた告知は 10.6.0 の ANNOUNCE メール です。 ↩︎
Maka 从 8 月开始也提出 Workhub 的概念，也就是由一个前端的 inbox bot 对所有需求进行分拣，交给不同的 Session 执行。 包括 Claude 的 Project，还有 Cursor 也有类似的设计，大家不约而同的走向了去 Session 的这个方向。 maka#3286
Your First Time-Range Query: A Practical Starting Point for Apache IoTDB > Start with one device, one measurement, and a clear interval to turn time-series history into an answerable question. Most time-series questions start with a window: what happened to this device between two points in time? An operator may want to inspect a temperature change during a shift. An engineer may need to review vibration before a maintenance event. A developer may be building a chart that shows how a signal moved over the last hour. In each case, “give me the data” is not specific enough. The query needs a subject, a measurement, and a time interval. Apache IoTDB supports time-oriented queries over IoT measurements. A time-range query is a small building block, but it is also a useful way to make a larger data question precise. Turn the question into a query scope Before writing SQL, identify three things: - Which asset or device? Narrow the query to the equipment relevant to the question. - Which measurement? Choose the signal that helps answer it, such as temperature or vibration. - Which time window? State both the start and end of the period you want to inspect. For example, “How did motor temperature behave during the last hour?” is more actionable than “Show me the motor data.” It identifies an asset, a measurement, and a bounded interval. If the purpose is troubleshooting, the interval might be the period before and after an event. If it is routine monitoring, it might match a shift or operating window. This framing also helps prevent accidental over-fetching. A query that has no clear time boundary may return more history than the chart, investigation, or application needs. A bounded range makes the intent easier to review and the result easier to handle. A basic Apache IoTDB time filter The following example uses the Tree model and an illustrative device path. Replace the path and measurement with names from your own schema: SELECT temperature FROM root.factory.line1.motor1 WHERE time >= now() - 1h This asks for temperature observations from the last hour for one device. The filter uses the time column to restrict the query window. For a historical investigation, use explicit start and end values appropriate to the target IoTDB release and the timestamp precision configured in the environment. The example is intentionally small. It does not aggregate the readings, interpolate missing points, or combine devices. Those are different analytical decisions and should be added only when the question requires them. Choose boundaries deliberately A time range is part of the meaning of a query. “The last hour” is useful for a live view, but a fixed interval can be easier to reproduce during an investigation. Whichever form you choose, decide whether the beginning and end are inclusive, and keep that choice consistent across adjacent queries. Boundary behavior matters when data is divided into windows. If one query ends at the same timestamp where the next begins and both include the boundary, a point can appear twice. If both exclude it, the point may be missed. A common design is to use a half-open interval: include the start, exclude the end. Verify that the syntax and application logic match the convention you choose. Also distinguish event time from arrival time. A record received during the selected hour may describe an observation from earlier. If the task is to understand equipment behavior, the query should follow the event timestamp preserved by the ingestion path, not silently substitute processing time. Keep the device and measurement scope readable Time-range queries are easier to debug when the path is explicit. A device-oriented path such as root.factory.line1.motor1 communicates which equipment the query addresses. A measurement name such as temperature states what is being retrieved. In an application, do not concatenate unchecked user input into a query. Validate device paths and measurement names against an allowed schema or metadata source. Validate the requested time bounds and reject ranges that exceed the product or application policy. These are application safeguards around query construction; they are not a replacement for database permissions. For a query over several devices, first define what the result should mean. Do you want separate histories for each asset, or do you want measurements aligned for comparison? The answer may affect how the application structures the request and presents the returned timestamps. Avoid combining paths simply because they share a chart. From a query result to an operational answer Returning rows is only the first step. The client should preserve timestamps when it converts results into chart points, API responses, or exported data. Sorting and timezone presentation should be consistent with the application’s timestamp policy. If the query returns no rows, the interface should distinguish “no observations in this interval” from “the device or measurement was not found.” For larger ranges, decide whether the user needs raw readings or a summary. A trend chart may need every point for a short period; a longer report may be better served by a time-window aggregation. Aggregation changes the question from “what values were recorded?” to “what summary describes each window?” That deserves a separate query design rather than an invisible transformation. Conclusion A time-range query is most useful when it expresses a concrete question. Select the device, choose the measurement, define the interval, and preserve the timestamp meaning through the client application. That gives operators and developers a clear starting point for retrieving and investigating IoT history. Apache IoTDB is an open-source time-series database for IoT workloads. Explore the project and its current query guidance in the Apache IoTDB GitHub repository. !Image description
Пишем высокопроизводительный rate-limiter на golang О чём статья Недавно я познакомился с golang и в качестве практики решил реализовать алгоритмы rate limiting'а из статьи c arxiv. В этой статье мы рассмотрим написание rate-limiter'а в виде отдельно стоящего сервиса и в виде middleware, используя in-memory кэш или Redis, добавим поддержку REST и gRPC, подключим Prometheus и поймём как добиться атомарности операций с минимальными задержками. Кому полезно Новичкам в golang или же людям, которые по какой-то причине также решили использовать собственные реализации rate-limiter'ов заместо уже существующих решений. Предисловие Указанная ранее статья покрывает 3 алгоритма rate-limiting'а: Token bucket Fixed window Rolling window Суть данных алгоритмов вы можете прочитать самостоятельно, я же сфокусируюсь на том, как написать данные алгоритмы таким образом, чтобы rate-limiter работал быстро и надёжно. Конфигурация Первое, с чего стоит начать, это конфигурация вашего limiter'а. В своей реализации я ставил следующие требования: Возможность переключаться между алгоритмами Поддержка in-memory кэша и Redis Метрики в Prometheus Поддержка REST и gRPC Из этих требований вытекает следующий файл конфигурации: store: "redis" use_algo: "token_bucket" metrics: use_prometheus: true port: 9091 algo_settings: capacity: 10 rate: 0.1 backends: in_memory: default_ttl: 30 # seconds eviction_time: 120 # seconds shards_count: 256 redis: addr: "redis:6379" password: "m1re1di77sss" db: 0 default_ttl: 1000sourcecraft.dev Рассмотрим его подробнее: store отвечает за используемое хранилище, может быть "redis" или же "in_memory" use_algo отвечает за используемый алгоритм metrics позволяет включать и отключать поддержку выгрузки и подсчёта метрик во время работы, а также за порт, на котором позже будет поднят мультиплексер для метрик algo_settings для каждого алгоритма разные, отвечает за настройки, которые будут использоваться алгоритмом backends - настройки каждого из хранилищ, в in_memory default_ttl отвечает за ttl ключей в хранилище, eviction_time отвечает за частоту очистки in-memory хранилища от протухших ключей, shard_count отвечает за то, на сколько частей in-memory кэш будет шардирован, число должно быть степенью 2, позже я объясню почему Конфигурация для middleware выглядит похожим образом: type InMemoryBackendSettings struct { DefaultTtl int64 yaml:"default_ttl" EvictionTime int64 yaml:"eviction_time" ShardsCount uint32 yaml:"shards_count" } type RedisBackendSettings struct { Addr string yaml:"addr" Password string yaml:"password" Db int yaml:"db" DefaultTtl int64 yaml:"default_ttl" } type AlgoSettings struct { Capacity *float64 yaml:"capacity" Rate *float64 yaml:"rate" WindowSize *int64 yaml:"window_size" MaxRequests *int64 yaml:"max_requests" }sourcecraft.dev Бэкенды для limiter'а Думаю, ни для кого не секрет, что Redis умеет выполнять Lua скрипты атомарно, что является идеальной функциональной возможностью для надёжного rate-limiting'а. В таком случае наш бэкенд limiter'а также должен уметь выполнять операции атомарно, отсюда вытекает интерфейс бэкенда: type InMemoryStoreInterface interface { Update(key string, newFn func() any, fn func(v any) error) error Close() error }sourcecraft.dev Update принимает на вход ключ, по которому мы хотим принять решение, а также 2 колбэка: newFn - функция создания нового объекта в хранилище, в случае если данный ключ в хранилище ещё не появлялся. Это позволяет избежать создания данного объекта в случае, если ключ уже содержится в хранилище. Конечно, можно было сначала проверять и только потом добавлять, но это добавляет лишней логики клиенту, использующему хранилище. fn - собственно функция, содержащая логику limiting'а In-memory кэш Наше хранилище под капотом будет использовать мапу, во избежание race condition'ов мы будем использовать mutex. Но ставить mutex на всю мапу слишком дорого, так как другие клиенты всё ещё должны пользоваться хранилищем. Отсюда вытекает необходимость в шардировании. Пока одна горутина выполняет логику по ключу x, мы всё ещё можем работать с ключом y: type shard struct { mu sync.Mutex entries map[string]*cacheEntry } type cacheEntry struct { entry any lastAccess int64 } type ShardedInMemoryStore struct { shards []*shard shardsCount uint32 expirationTime int64 ticker *time.Ticker done chan any once sync.Once } func NewCache( expirationTime int64, evictionInterval time.Duration, shardCount uint32, ) *ShardedInMemoryStore { ticker := time.NewTicker(evictionInterval) cache := &ShardedInMemoryStore{ shards: make([]*shard, shardCount), shardsCount: shardCount, expirationTime: expirationTime, done: make(chan any), ticker: ticker, } for i := range shardCount { cache.shards[i] = &shard{entries: make(map[string]*cacheEntry)} } go func() { defer ticker.Stop() for { select { case <-cache.done: return case <-ticker.C: cache.handleEviction() } } }() return cache } sourcecraft.dev Также удаление протухших ключей теперь не будет лочить всю мапу, а только один конкретный шард: func (cache *ShardedInMemoryStore) handleEviction() { currentTime := time.Now().Unix() for _, shard := range cache.shards { shard.mu.Lock() for key, value := range shard.entries { if currentTime-cache.expirationTime > value.lastAccess { delete(shard.entries, key) } } shard.mu.Unlock() } }sourcecraft.dev А также ранее упомянутый Update: func (cache *ShardedInMemoryStore) Update(key string, createFn func() any, fn func(v any) error) error { shard := cache.getShard(key) shard.mu.Lock() defer shard.mu.Unlock() now := time.Now().Unix() value, present := shard.entries[key] if !present { value = &cacheEntry{entry: createFn(), lastAccess: now} shard.entries[key] = value } value.lastAccess = now return fn(value.entry) }sourcecraft.dev Именно здесь реализуется столь необходимая атомарность, мы лочим mutex на шарде и выполняем операцию, не блокируя другие шарды и не допуская race condition'ов. Так почему количество шардов должно быть степенью двойки? Для того, чтобы распределять клиентские ключи более-менее равномерно по шардам нужно использоват функцию хэширования, которая не только будет раскладывать ключи равномерно, но и будет быстрой. Под такую функцию подходит Fowler–Noll–Vo. А чтобы после этого быстро замапить хэш на шард можно воспользоваться трюком. Вкратце можно очень быстро искать остатки от деления на степень двойки просто используя формулу: hash & (shardsCount - 1): func (cache *ShardedInMemoryStore) getShard(key string) *shard { h := uint32(2166136261) for i := 0; i < len(key); i++ { h ^= uint32(key[i]) h *= 16777619 } return cache.shards[h&(cache.shardsCount-1)] }sourcecraft.dev Redis кэш С Redis всё проще: package cache import ( "context" "github.com/redis/go-redis/v9" ) type RedisStoreInterface interface { RunLuaScript(ctx context.Context, script *redis.Script, keys []string, args ...any) (bool, int64, error) DefaultTtl() int64 }sourcecraft.dev Передаём контекст, скрипт и массив ключей, в данном случае массив ключей это требование библиотеки для работы с Redis в Golang. Массив всегда будет состоять из одного элемента. DefaultTtl - это getter, который используется для token bucket алгоритма, чтобы удалением ключа при использовании Redis занимался сам Redis, а не приложение. package cache import ( "context" "fmt" "github.com/redis/go-redis/v9" ) type RedisStore struct { client *redis.Client defaultTtl int64 } func NewRedisStore(client *redis.Client, defaultTtl int64) *RedisStore { return &RedisStore{client: client, defaultTtl: defaultTtl} } // RunLuaScript runs a Lua script that returns {allowed(0|1), retryAfter}. func (r *RedisStore) RunLuaScript( ctx context.Context, script *redis.Script, keys []string, args ...any, ) (bool, int64, error) { res, err := script.Run(ctx, r.client, keys, args...).Int64Slice() if err != nil { return false, 0, err } if len(res) != 2 { return false, 0, fmt.Errorf("unexpected length of script output") } return res[0] == 1, res[1], nil } func (r *RedisStore) DefaultTtl() int64 { return r.defaultTtl }sourcecraft.dev Используем созданные хранилища Теперь рассмотрим функцию запуска limiter'а: package main import ( "context" "fmt" "log" "net" "net/http" "os" "os/signal" "qwrttqr-rate-limiter/core/config" "qwrttqr-rate-limiter/core/server/algos" "qwrttqr-rate-limiter/core/server/facade" "qwrttqr-rate-limiter/core/server/limiting" "qwrttqr-rate-limiter/core/server/metrics" ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1" "strconv" "syscall" "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" "google.golang.org/grpc" ) func main() { if err := run(); err != nil { log.Fatal(err) } } func run() error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() cfg, err := config.ReadConfig() if err != nil { return fmt.Errorf("failed to read config: %w", err) } rl, err := limiting.NewRateLimiter(cfg) if err != nil { return fmt.Errorf("create rate limiter: %w", err) } defer rl.Close() var limiter algos.Limiter = rl errCh := make(chan error, 3) shutdownMetrics := func(ctx context.Context) {} if cfg.Metrics.UsePrometheus { reg := prometheus.NewRegistry() rec := metrics.NewPromRecorder(reg) limiter = metrics.Instrument(limiter, rec) metricsMux := http.NewServeMux() metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{})) metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second} go func() { errCh <- metricsSrv.ListenAndServe() }() log.Printf("metrics available on %s", metricsSrv.Addr) shutdownMetrics = func(ctx context.Context) { _ = metricsSrv.Shutdown(ctx) } } limitingMux := http.NewServeMux() limitingMux.HandleFunc("GET /health", facade.HealthHandler) limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter)) limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second} grpcSrv := grpc.NewServer() ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter)) tcpLis, err := net.Listen("tcp", ":8081") if err != nil { return fmt.Errorf("listen grpc: %w", err) } go func() { errCh <- limitingSrv.ListenAndServe() }() log.Printf("listening http on %s", limitingSrv.Addr) go func() { errCh <- grpcSrv.Serve(tcpLis) }() log.Printf("listening grpc on %s", tcpLis.Addr()) select { case err := <-errCh: return err case <-ctx.Done(): } shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() grpcDone := make(chan struct{}) go func() { grpcSrv.GracefulStop(); close(grpcDone) }() select { case <-grpcDone: case <-shutdownCtx.Done(): grpcSrv.Stop() } shutdownMetrics(shutdownCtx) return limitingSrv.Shutdown(shutdownCtx) } sourcecraft.dev Давайте по порядку: ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop()sourcecraft.dev Создаём shutdown context, который срабатывает на нажатие Ctrl-C либо при получении syscall.SIGTERM (сигнал, посылаемый родительским процессом на завершение нашего процесса, например когда тушим докер-контейнер). defer stop() нужен для того, чтобы прекратить обрабатывать данные сигналы, сразу перед возвратом из run. cfg, err := config.ReadConfig() if err != nil { return fmt.Errorf("failed to read config: %w", err) } rl, err := limiting.NewRateLimiter(cfg) if err != nil { return fmt.Errorf("create rate limiter: %w", err) } defer rl.Close()sourcecraft.dev Далее читаем конфиг и создаём сам limiter, у limiter'а есть функция Close, нужная для закрытия соединения с Redis. в in-memory кэше она просто закрывает канал, что останавливает удаление ключей из кэша. var limiter algos.Limiter = rl errCh := make(chan error, 3)sourcecraft.dev Следующие строки позволяют создать указатель с типом интерфейса на limiter: package algos import ( "context" ) type Request struct { Key string RequiredTokens float64 } type Decision struct { Allowed bool // RetryAfter is seconds RetryAfter int64 } type Limiter interface { Allow(ctx context.Context, req Request) (Decision, error) }sourcecraft.dev Так как NewRateLimiter возвращает не интерфейс, а указатель на структуру, поддерживающую Close и Allow: type RateLimiter struct { algos.Limiter io.Closer } func NewRateLimiter(cfg config.StandaloneServiceConfiguration) (_ *RateLimiter, err error) { var ( mem cache.InMemoryStoreInterface rdb *redis.Client closer io.Closer ) switch cfg.Store { case "in_memory": cacheInstance := cache.NewCache( cfg.Backends.InMemory.DefaultTtl, time.Duration(cfg.Backends.InMemory.EvictionTime)*time.Second, cfg.Backends.InMemory.ShardsCount, ) mem, closer = cacheInstance, cacheInstance case "redis": redisClient := redis.NewClient(&redis.Options{ Addr: cfg.Backends.Redis.Addr, Password: cfg.Backends.Redis.Password, DB: cfg.Backends.Redis.Db, }) rdb, closer = redisClient, redisClient default: return nil, fmt.Errorf("unsupported store: %q", cfg.Store) } defer func() { if err != nil { _ = closer.Close() } }() var limiter algos.Limiter switch cfg.UsedAlgo { case "token_bucket": if err := algos.ValidateTokenBucketConfig(cfg.AlgoSettings); err != nil { return nil, err } store, err := algos.NewTokenBucketStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl) if err != nil { return nil, err } limiter = &algos.TokenBucketLimiter{ Capacity: *cfg.AlgoSettings.Capacity, Rate: *cfg.AlgoSettings.Rate, Store: store, Now: realClock, } case "fixed_window": if err := algos.ValidateFixedWindowConfig(cfg.AlgoSettings); err != nil { return nil, err } store, err := algos.NewFixedWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl) if err != nil { return nil, err } limiter = &algos.FixedWindowLimiter{ MaxRequests: *cfg.AlgoSettings.MaxRequests, WindowSize: *cfg.AlgoSettings.WindowSize, Store: store, Now: realClock, } case "rolling_window": if err := algos.ValidateRollingWindowConfig(cfg.AlgoSettings); err != nil { return nil, err } store, err := algos.NewRollingWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl) if err != nil { return nil, err } limiter = &algos.RollingWindowLimiter{ MaxRequests: *cfg.AlgoSettings.MaxRequests, WindowSize: *cfg.AlgoSettings.WindowSize, Store: store, Now: realClock, } default: return nil, fmt.Errorf("unsupported rate limiting algorithm: %s", cfg.UsedAlgo) } return &RateLimiter{Limiter: limiter, Closer: closer}, nil }sourcecraft.dev Возвращаясь обратно в run: shutdownMetrics := func(ctx context.Context) {} if cfg.Metrics.UsePrometheus { reg := prometheus.NewRegistry() rec := metrics.NewPromRecorder(reg) limiter = metrics.Instrument(limiter, rec) metricsMux := http.NewServeMux() metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{})) metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second} go func() { errCh <- metricsSrv.ListenAndServe() }() log.Printf("metrics available on %s", metricsSrv.Addr) shutdownMetrics = func(ctx context.Context) { _ = metricsSrv.Shutdown(ctx) } }sourcecraft.dev Мы помечали rl, как структуру, реализующую интерфейс как раз ради поддержки метрик здесь. О том, как метрики работают я расскажу позже, сейчас достаточно понимания того, что мы создаём специальный HTTP сервер, с которого Prometheus будет забирать метрики. Для подсчёта данных метрик мы используем Registry, а в функции instrument мы оборачиваем limiter в мидлварь, которая будет обновлять счётчики. Функция shutdownMetrics обновляется в зависимости от того, включаем ли у себя метрики или нет, это нужно, чтобы сделать graceful shutdown данного сервера. limitingMux := http.NewServeMux() limitingMux.HandleFunc("GET /health", facade.HealthHandler) limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter)) limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second} grpcSrv := grpc.NewServer() ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter)) tcpLis, err := net.Listen("tcp", ":8081") if err != nil { return fmt.Errorf("listen grpc: %w", err) } go func() { errCh <- limitingSrv.ListenAndServe() }() log.Printf("listening http on %s", limitingSrv.Addr) go func() { errCh <- grpcSrv.Serve(tcpLis) }() log.Printf("listening grpc on %s", tcpLis.Addr()) select { case err := <-errCh: return err case <-ctx.Done(): }sourcecraft.dev После этого создаём HTTP и gRPC сервера и ждём в select на ранее созданном NotifyContext, до момента пока не придёт сигнал на завершение процесса (os.Interrupt, syscall.SIGTERM) или ошибка. shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() grpcDone := make(chan struct{}) go func() { grpcSrv.GracefulStop(); close(grpcDone) }() select { case <-grpcDone: case <-shutdownCtx.Done(): grpcSrv.Stop() } shutdownMetrics(shutdownCtx) return limitingSrv.Shutdown(shutdownCtx)sourcecraft.dev В случае, если к нам пришла ошибка или же сигнал на завершение процесса, создаём контекст с таймаутом и просим все наши сервера выключиться. В особенности это касается gRPC сервера, он самостоятельно поддерживает GracefulStop, но если он не успеет сделать это за 10 секунд, мы выключим его принудительно. Используем ранее созданные бэкенды Помните, что мы создали Redis и InMemory бэкенды для наших алгоритмов, теперь можно посмотреть, как мы пользуемся ими, рассмотрим пример с Rolling Window: type RollingWindowStore interface { CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error) } type RollingWindowLimiter struct { WindowSize int64 MaxRequests int64 Store RollingWindowStore Now func() int64 } func (rwl *RollingWindowLimiter) Configure() { if rwl.Now == nil { rwl.Now = func() int64 { return time.Now().Unix() } } } func ValidateRollingWindowConfig(cfg config.AlgoSettings) error { if *cfg.MaxRequests <= 0 { return fmt.Errorf("the max requests param should be strictly greater then 0") } return CheckRequiredFields(cfg, []string{"window_size", "max_requests"}) } type InMemoryRollingWindowStore struct { Backend cache.InMemoryStoreInterface } type RedisRollingWindowStore struct { Backend cache.RedisStoreInterface } func NewRollingWindowStore(store string, cacheInstance cache.InMemoryStoreInterface, redisClient *redis.Client, redisDefaultKeyTTL int64) (RollingWindowStore, error) { switch store { case "in_memory": return &InMemoryRollingWindowStore{Backend: cacheInstance}, nil case "redis": return &RedisRollingWindowStore{Backend: cache.NewRedisStore(redisClient, redisDefaultKeyTTL)}, nil default: return nil, fmt.Errorf("unsupported store: %s", store) } }sourcecraft.dev Ещё один интерфейс(RollingWindowStore), нужен для генерализации работы с бэкендом, таким образом, сам алгоритм будет просто вызывать функцию CheckAndIncrement, передавая параметры, а бэкенд будет выполнять её в атомарном варианте. func (rwl *RollingWindowLimiter) Configure() { if rwl.Now == nil { rwl.Now = func() int64 { return time.Now().Unix() } } }sourcecraft.dev Данная функция нужна для тестов, так как в тестах придётся мокать функцию получения текущего времени. type RollingWindowState struct { Timestamps []int64 } func (rws *RollingWindowState) check( windowSize, maxRequests, now int64, ) (bool, int64) { windowStart := now - windowSize i := 0 for ; i < len(rws.Timestamps); i++ { if rws.Timestamps[i] > windowStart { break } } if i > 0 { copy(rws.Timestamps, rws.Timestamps[i:]) rws.Timestamps = rws.Timestamps[:len(rws.Timestamps)-i] } if int64(len(rws.Timestamps)) < maxRequests { rws.Timestamps = append(rws.Timestamps, now) return true, 0 } retryAfter := rws.Timestamps[0] - windowStart return false, retryAfter }sourcecraft.dev Сам алгоритм устроен несложно, мы просто отслеживаем, какие временные метки уже вышли за временное окно, и если оставшихся меток меньше, чем лимит запросов, клиент может выполнить очередной запрос. Иначе нужно ответить, через какое количество секунд освободится хотя бы одно место. Далее подвязываем данную функцию к бэкенду: func (store *InMemoryRollingWindowStore) CheckAndIncrement( _ context.Context, key string, windowSize, maxRequests, now int64, ) (allowed bool, retryAfter int64, err error) { err = store.Backend.Update( key, func() any { return &RollingWindowState{Timestamps: make([]int64, 0, 8)} }, func(v any) error { st, ok := v.(*RollingWindowState) if !ok { return fmt.Errorf("unexpected state type %T for key %q", v, key) } allowed, retryAfter = st.check(windowSize, maxRequests, now) return nil }) return allowed, retryAfter, err }sourcecraft.dev То же самое в Redis: var rollingWindowScript = redis.NewScript(` local requestId = ARGV[1] local windowSize = tonumber(ARGV[2]) local maxRequests = tonumber(ARGV[3]) local now = tonumber(ARGV[4]) local windowStart = now - windowSize redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", windowStart) local count = redis.call("ZCARD", KEYS[1]) if count < maxRequests then redis.call("ZADD", KEYS[1], now, requestId) redis.call("EXPIRE", KEYS[1], windowSize) return {1, 0} else local oldest = redis.call("ZRANGE", KEYS[1], 0, 0, "WITHSCORES") local retryAfter = 0 if oldest and oldest[2] then retryAfter = tonumber(oldest[2]) - windowStart end return {0, retryAfter} end `) func (s *RedisRollingWindowStore) CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error) { requestId := uuid.New().String() return s.Backend.RunLuaScript(ctx, rollingWindowScript, []string{key}, requestId, windowSize, maxRequests, now) }sourcecraft.dev В Redis реализации стоит обратить внимание на то, что мы генерируемrequestId, всё потому, что мы храним наши элементы в множестве, а в множестве могут храниться только уникальные элементы. Но как тогда быть в ситуации, когда 2 и более запросов прилетают в одну секунду? Для этого и создан requestId, он добавляет уникальность каждому запросу, поэтому если в одну секунду прилетит 1000 запросов, они все посчитаются как разные запросы. А что с метриками? Как я ранее говорил, rate-limiter поддерживает подсчёт метрик. Начнём с функции instrument, о которой говорилось ранее: package metrics import ( "context" "qwrttqr-rate-limiter/core/server/algos" "time" ) type instrumented struct { next algos.Limiter rec Recorder } func Instrument(next algos.Limiter, rec Recorder) algos.Limiter { return &instrumented{next: next, rec: rec} } func (i *instrumented) Allow(ctx context.Context, req algos.Request) (algos.Decision, error) { start := time.Now() dec, err := i.next.Allow(ctx, req) elapsed := time.Since(start) switch { case err != nil: i.rec.ObserveDecision(Errored, elapsed) case dec.Allowed: i.rec.ObserveDecision(Allowed, elapsed) default: i.rec.ObserveDecision(Denied, elapsed) } return dec, err }sourcecraft.dev Данная функция просто возвращает структуру, которая обладает методом Allow, таким образом мы подменяем наш оригинальный limiter обёрткой, которая поддерживает подсчёт метрик. Recorder реализован так: package metrics import "time" type Result uint8 const ( Allowed Result = iota Denied Errored NumResults ) type Recorder interface { ObserveDecision(r Result, elapsed time.Duration) }sourcecraft.dev И конкретная реализация данного интерфейса под Prometheus выглядит так: package metrics import ( "time" "github.com/prometheus/client_golang/prometheus" ) type PromRecorder struct { decisions [NumResults]prometheus.Counter durationHist [NumResults]prometheus.Observer } func NewPromRecorder(reg prometheus.Registerer) *PromRecorder { counterVec := prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "ratelimiter", Name: "decisions_total", Help: "Each decision count by its type.", }, []string{"result"}) reqDurHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "ratelimiter", Name: "decision_duration_seconds", Help: "Execution time of one decision in seconds.", Buckets: []float64{0.00005, 0.0001, 0.0005, 0.001, 0.1, 0.5, 1, 2}, }, []string{"result"}) reg.MustRegister(counterVec) reg.MustRegister(reqDurHistVec) p := &PromRecorder{} p.decisions[Allowed] = counterVec.WithLabelValues("allowed") p.decisions[Denied] = counterVec.WithLabelValues("denied") p.decisions[Errored] = counterVec.WithLabelValues("errored") p.durationHist[Allowed] = reqDurHistVec.WithLabelValues("allowed") p.durationHist[Denied] = reqDurHistVec.WithLabelValues("denied") p.durationHist[Errored] = reqDurHistVec.WithLabelValues("errored") return p } func (p *PromRecorder) ObserveDecision(r Result, dur time.Duration) { p.decisions[r].Inc() p.durationHist[r].Observe(dur.Seconds()) }sourcecraft.dev Здесь мы создаём счётчики и гистограммы под каждый результат rate-limiter'а и реализуем ObserveDecision, который их обновляет. gRPC и REST gRPC Пишем .proto файл: syntax = "proto3"; package ratelimiter.v1; option go_package = "qwrttqr-rate-limiter/gen/ratelimiter/v1;ratelimiterv1"; import "google/protobuf/duration.proto"; service RateLimiter { rpc Allow(AllowRequest) returns (AllowResponse); } message AllowRequest { string key = 1; double tokens = 2; } message AllowResponse { bool allowed = 1; google.protobuf.Duration retryAfter = 2; }sourcecraft.dev Вызываем protoc и остаётся просто реализовать полученные интерфейсы: protoc -I proto --go_out=. --go_opt=module=qwrttqr-rate-limiter --go-grpc_out=. --go-grpc_opt=module=qwrttqr-rate-limiter ratelimiter/v1/ratelimiter.protosourcecraft.dev package facade import ( "context" "qwrttqr-rate-limiter/core/server/algos" ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1" "time" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/durationpb" ) type GRPCServer struct { ratelimiterv1.UnimplementedRateLimiterServer l algos.Limiter } func NewGRPCServer(l algos.Limiter) *GRPCServer { return &GRPCServer{l: l} } func (s *GRPCServer) Allow(ctx context.Context, in *ratelimiterv1.AllowRequest) (*ratelimiterv1.AllowResponse, error) { tokens := in.GetTokens() if tokens == 0.0 { tokens = 1.0 } req, err := NewRequest(in.GetKey(), tokens) if err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) } d, err := s.l.Allow(ctx, req) if err != nil { return nil, status.Error(codes.Unavailable, "rate limiter error") } return &ratelimiterv1.AllowResponse{ Allowed: d.Allowed, RetryAfter: durationpb.New(time.Duration(d.RetryAfter) * time.Second), }, nil }sourcecraft.dev REST Для REST всё совсем просто: package facade import ( "net/http" "qwrttqr-rate-limiter/core/server/algos" "strconv" ) func HealthHandler(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("ok")) } func LimitHandler(l algos.Limiter) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { key := r.Header.Get("X-Client-Key") tokens := 1.0 if s := r.Header.Get("X-Required-Tokens"); s != "" { n, err := strconv.ParseFloat(s, 64) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } tokens = n } req, err := NewRequest(key, tokens) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } d, err := l.Allow(r.Context(), req) switch { case err != nil: http.Error(w, "rate limiter error", http.StatusInternalServerError) // store failure case !d.Allowed: w.Header().Set("Retry-After", strconv.FormatInt(d.RetryAfter, 10)) http.Error(w, "rate limit exceeded", http.StatusTooManyRequests) } } }sourcecraft.dev Middleware Для реализации поддержки middleware можно использовать такую схему: Приложение создаёт себе бэкенд, используя Redis или оставаясь на in-memory кэше Используя созданный бэкенд создаём себе rate-limiter, с нужным алгоритмом package middleware import ( "qwrttqr-rate-limiter/core/config" "qwrttqr-rate-limiter/core/server/algos" "qwrttqr-rate-limiter/core/server/cache" "time" "github.com/redis/go-redis/v9" ) var realClock = func() int64 { return time.Now().Unix() } type Backend interface { TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) } type memoryBackend struct { mem cache.InMemoryStoreInterface } func NewInMemoryBackend(cfg config.InMemoryBackendSettings) Backend { return &memoryBackend{mem: cache.NewCache( cfg.DefaultTtl, time.Duration(cfg.EvictionTime)*time.Second, cfg.ShardsCount), } } type redisBackend struct { redis cache.RedisStoreInterface } func NewRedisBackend(client *redis.Client, redisDefaultKeyTTL int64) Backend { return &redisBackend{redis: cache.NewRedisStore(client, redisDefaultKeyTTL)} } func newTokenBucket(cfg config.AlgoSettings, store algos.TokenBucketStore) (algos.Limiter, error) { if err := algos.ValidateTokenBucketConfig(cfg); err != nil { return nil, err } return &algos.TokenBucketLimiter{ Capacity: *cfg.Capacity, Rate: *cfg.Rate, Store: store, Now: realClock, }, nil } func newFixedWindow(cfg config.AlgoSettings, store algos.FixedWindowStore) (algos.Limiter, error) { if err := algos.ValidateFixedWindowConfig(cfg); err != nil { return nil, err } return &algos.FixedWindowLimiter{ WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock, }, nil } func newRollingWindow(cfg config.AlgoSettings, store algos.RollingWindowStore) (algos.Limiter, error) { if err := algos.ValidateRollingWindowConfig(cfg); err != nil { return nil, err } return &algos.RollingWindowLimiter{ WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock, }, nil } func (b *memoryBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) { return newTokenBucket(cfg, &algos.InMemoryTokenBucketStore{Backend: b.mem}) } func (b *memoryBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newFixedWindow(cfg, &algos.InMemoryFixedWindowStore{Backend: b.mem}) } func (b *memoryBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newRollingWindow(cfg, &algos.InMemoryRollingWindowStore{Backend: b.mem}) } func (b *redisBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) { return newTokenBucket(cfg, &algos.RedisTokenBucketStore{Backend: b.redis}) } func (b *redisBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newFixedWindow(cfg, &algos.RedisFixedWindowStore{Store: b.redis}) } func (b *redisBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newRollingWindow(cfg, &algos.RedisRollingWindowStore{Backend: b.redis}) }sourcecraft.dev Далее остаётся написать функцию, создающую непосредственно саму middleware, а именно функцию возвращающую тип, поддерживающий serveHTTP метод: package middleware import ( "log" "net/http" "qwrttqr-rate-limiter/core/server/algos" "qwrttqr-rate-limiter/core/server/facade" "strconv" ) func NewHTTPMiddleware( algo algos.Limiter, keyDer HTTPKeyDerivationFunction, costDer HTTPCostDerivationFunction, failOpen bool, ) func(http.Handler) http.Handler { if algo == nil || keyDer == nil { panic("middleware: limiter and key function are required") } return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fail := func(stage string, err error) { log.Printf("limiter error on %s: %v", stage, err) if failOpen { next.ServeHTTP(w, r) return } http.Error(w, "service unavailable", http.StatusServiceUnavailable) } tokens := 1.0 if costDer != nil { t, err := costDer(r) if err != nil { fail("cost derivation", err) return } tokens = t } key, err := keyDer(r) if err != nil { fail("key derivation", err) return } req, err := facade.NewRequest(key, tokens) if err != nil { fail("request creation", err) return } decision, err := algo.Allow(r.Context(), req) if err != nil { fail("allow", err) return } if !decision.Allowed { w.Header().Set("Retry-After", strconv.FormatInt(decision.RetryAfter, 10)) http.Error(w, "rate limit exceeded", http.StatusTooManyRequests) return } next.ServeHTTP(w, r) }) } }sourcecraft.dev С gRPC необходимо реализовать 2 вида interceptor'а - унарный и потоковый: func NewGRPCUnaryInterceptor( algo algos.Limiter, keyDer GRPCKeyDerivationFunction, costDer GRPCCostDerivationFunction, failOpen bool, ) grpc.UnaryServerInterceptor { check := newGRPCCheck(algo, keyDer, costDer, failOpen) return func( ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (any, error) { if err := check(ctx); err != nil { return nil, err } return handler(ctx, req) } } func NewGRPCStreamInterceptor( algo algos.Limiter, keyDer GRPCKeyDerivationFunction, costDer GRPCCostDerivationFunction, failOpen bool, ) grpc.StreamServerInterceptor { check := newGRPCCheck(algo, keyDer, costDer, failOpen) return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { if err := check(ss.Context()); err != nil { return err } return handler(srv, ss) } }sourcecraft.dev И check: func newGRPCCheck( algo algos.Limiter, keyDer GRPCKeyDerivationFunction, costDer GRPCCostDerivationFunction, failOpen bool, ) func(ctx context.Context) error { if algo == nil || keyDer == nil { panic("middleware: limiter and key function are required") } return func(ctx context.Context) error { fail := func(stage string, err error) error { log.Printf("limiter error on %s: %v", stage, err) if failOpen { return nil } return status.Error(codes.Unavailable, "service unavailable") } tokens := 1.0 if costDer != nil { t, err := costDer(ctx) if err != nil { return fail("cost derivation", err) } tokens = t } key, err := keyDer(ctx) if err != nil { return fail("key derivation", err) } req, err := facade.NewRequest(key, tokens) if err != nil { return fail("request creation", err) } decision, err := algo.Allow(ctx, req) if err != nil { return fail("allow", err) } if !decision.Allowed { md := metadata.Pairs("retry-after", strconv.FormatInt(decision.RetryAfter, 10)) _ = grpc.SetTrailer(ctx, md) return status.Error(codes.ResourceExhausted, "rate limit exceeded") } return nil } }sourcecraft.dev Как вы могли заметить, везде можно передать функции извлечение ключа и стоимости запроса. Это нужно так как в режиме middleware вы чаще всего будете писать сервисы, которые смотрят наружу, а значит будут принимать запросы от недоверенных клиентов. Итог Статья получилась достаточно объёмной, надеюсь она будет полезна ребятам, которые только начинают учить Golang) исходник
Unstructured; Supabase: Postgres development platform; yt-dlp | GitHub Curated Picks | 2026-10-23 GitHub Curated Picks reviews 14 source-verified repositories selected for developer usefulness, not popularity position. Each pick is checked against its GitHub repository, README, release evidence, and recent commit; pick numbers are editorial order. The compact card keeps rank, signal, why it matters, and next check together while the ledger preserves direct URLs and release evidence. Treat the list as an educational review surface, not software endorsement, and verify license and maintenance before adoption. Chapters: 00:00 2026-10-23 | GitHub Curated Picks 00:05 Pick #1 Unstructured 00:30 Pick #2 Supabase 00:44 Pick #3 yt-dlp 01:01 Pick #4 Langchain 01:27 Pick #5 Awesome Mac 01:42 Pick #6 Trivy 01:56 Pick #7 Vite 02:14 Pick #8 Jellyfin 02:32 Pick #9 Black 02:51 Pick #10 Coolify 03:13 Pick #11 Doris 03:24 Pick #12 Onnxruntime 03:36 Pick #13 Fluentui 03:49 Pick #14 Server Source ledger: github.com/trending api.github.com github.com/Unstructured-IO/unstructured/blob/main… github.com/Unstructured-IO/unstructured/commit/49… github.com/Unstructured-IO/unstructured/releases/… api.github.com github.com/supabase/supabase/blob/master/README.md github.com/supabase/supabase/commit/2193f2acdf8ef… api.github.com github.com/yt-dlp/yt-dlp/blob/master/README.md github.com/yt-dlp/yt-dlp/commit/9b58cb33d0cd3e823… api.github.com langchain-ai/langchain github.com/langchain-ai/langchain/commit/b84fa61e… github.com/langchain-ai/langchain/releases/tag/la… api.github.com github.com/jaywcjlove/awesome-mac/blob/master/REA… github.com/jaywcjlove/awesome-mac/commit/965f263f… api.github.com github.com/aquasecurity/trivy/blob/main/README.md github.com/aquasecurity/trivy/commit/0231990f702f… api.github.com github.com/vitejs/vite/blob/main/README.md github.com/vitejs/vite/commit/a236886026c3a5de068… api.github.com github.com/jellyfin/jellyfin/blob/master/README.md github.com/jellyfin/jellyfin/releases/tag/v12.3 api.github.com github.com/psf/black/blob/main/README.md github.com/psf/black/commit/f33c3e3c66263d4a3a0e1… api.github.com coollabsio/coolify github.com/coollabsio/coolify/commit/0b2e931a4258… api.github.com github.com/apache/doris/blob/master/README.md github.com/apache/doris/commit/6e5a8929fbaba2ed2a… api.github.com microsoft/onnxruntime github.com/microsoft/onnxruntime/commit/fe7700a7e… api.github.com microsoft/fluentui github.com/microsoft/fluentui/commit/ea1a99bf2bc5… api.github.com nextcloud/server github.com/nextcloud/server/commit/fc9b07d9591f1f… Unstructured-IO/unstructured supabase/supabase yt-dlp/yt-dlp jaywcjlove/awesome-mac aquasecurity/trivy vitejs/vite jellyfin/jellyfin psf/black apache/doris Informational and educational purposes only. Not investment advice, product endorsement, or professional guidance.
Apache Hop - 07 - Version control Version control with Git Give the project folder a Git history from Hop GUI. The first commit adds the pipeline and the workflow from the default project. A later commit changes the order id sequence from 1000 to 50000, and the visual diff shows both values side by side. Environment files and passwords stay out of the repository. Part of the Apache Hop getting started series. hop.apache.org Working with Git: hop.apache.org Git commit perspective: hop.apache.org Git perspective: hop.apache.org Issue: hop#9064 Chapters 0:00 Welcome 0:26 Why keep a history 0:56 When Git is on 1:37 Check out a branch 2:06 Commit 2:43 History 3:13 Visual diff [de-DE] Apache Hop - 07 - Version control Versionsverwaltung mit Git Gib dem project-Ordner aus Hop GUI eine Git-Geschichte. Der erste Commit fügt die Pipeline und den Workflow aus dem default project hinzu. Ein späterer Commit ändert die order-id-Folge von 1000 auf 50000, und der visuelle Vergleich zeigt beide Werte nebeneinander. environment-Dateien und Passwörter bleiben außerhalb des Repository. Teil der Einführungsserie zu Apache Hop. hop.apache.org Arbeiten mit Git: hop.apache.org Perspektive Git commit: hop.apache.org Perspektive Git: hop.apache.org Ticket: hop#9064 Kapitel 0:00 Willkommen 0:26 Warum eine Geschichte 0:56 Wenn Git an ist 1:37 Einen Branch auschecken 2:06 Commit 2:43 Geschichte 3:13 Visueller Vergleich [es-ES] Apache Hop - 07 - Version control Control de versiones con Git Dale al project una historia de Git desde Hop GUI. El primer commit añade el Pipeline y el Workflow del default project. Un commit posterior cambia la secuencia order id de 1000 a 50000, y la diferencia visual muestra los dos valores lado a lado. Los archivos de environment y las contraseñas se quedan fuera del repositorio. Parte de la serie de introducción a Apache Hop. hop.apache.org Trabajar con Git: hop.apache.org Perspectiva Git commit: hop.apache.org Perspectiva Git: hop.apache.org Incidencia: hop#9064 Capítulos 0:00 Bienvenida 0:26 Por qué guardar una historia 0:56 Cuando Git está activo 1:37 Hacer checkout de una rama 2:06 Commit 2:43 Historia 3:13 Diferencia visual [fr-FR] Apache Hop - 07 - Version control Contrôle de version avec Git Donnez au dossier du project un historique Git depuis Hop GUI. Le premier commit ajoute le Pipeline et le Workflow du default project. Un commit suivant change la séquence order id de 1000 à 50000, et le diff visuel montre les deux valeurs côte à côte. Les fichiers d'environment et les mots de passe restent hors du dépôt. Fait partie de la série de prise en main d'Apache Hop. hop.apache.org Travailler avec Git: hop.apache.org Perspective Git commit: hop.apache.org Perspective Git: hop.apache.org Ticket: hop#9064 Chapitres 0:00 Bienvenue 0:26 Pourquoi garder un historique 0:56 Quand Git est actif 1:37 Faire le checkout d'une branche 2:06 Commit 2:43 Historique 3:13 Diff visuel [it] Apache Hop - 07 - Version control Controllo di versione con Git Dai alla cartella del project una storia Git da Hop GUI. Il primo commit aggiunge il Pipeline e il Workflow del default project. Un commit successivo cambia la sequenza order id da 1000 a 50000, e il confronto visivo mostra i due valori uno accanto all'altro. I file di environment e le password restano fuori dal repository. Parte della serie introduttiva ad Apache Hop. hop.apache.org Lavorare con Git: hop.apache.org Prospettiva Git commit: hop.apache.org Prospettiva Git: hop.apache.org Segnalazione: hop#9064 Capitoli 0:00 Benvenuto 0:26 Perché tenere una storia 0:56 Quando Git è attivo 1:37 Fare il checkout di un ramo 2:06 Commit 2:43 Storia 3:13 Confronto visivo [pt-PT] Apache Hop - 07 - Version control Controlo de versões com Git Dá à pasta do project uma história Git a partir do Hop GUI. O primeiro commit adiciona o Pipeline e o Workflow do default project. Um commit posterior muda a sequência order id de 1000 para 50000, e a diferença visual mostra os dois valores lado a lado. Os ficheiros de environment e as palavras-passe ficam fora do repositório. Parte da série de introdução ao Apache Hop. hop.apache.org Trabalhar com Git: hop.apache.org Perspetiva Git commit: hop.apache.org Perspetiva Git: hop.apache.org Issue: hop#9064 Capítulos 0:00 Boas-vindas 0:26 Porque guardar uma história 0:56 Quando o Git está ativo 1:37 Fazer checkout de um ramo 2:06 Commit 2:43 História 3:13 Diferença visual [zh-Hans] Apache Hop - 07 - Version control 用 Git 做版本控制 在 Hop GUI 里给 project 文件夹加上 Git 历史。第一次提交加入 default project 里的 Pipeline 和 Workflow。之后的一次提交把 order id 序列从 1000 改成 50000，可视化差异并排显示这两个值。environment 文件和密码留在仓库之外。 Apache Hop 入门系列的一部分。 hop.apache.org 使用 Git: hop.apache.org Git commit 透视图: hop.apache.org Git 透视图: hop.apache.org 议题: hop#9064 章节 0:00 欢迎 0:26 为什么要有历史 0:56 Git 启用时 1:37 检出一个分支 2:06 提交 2:43 历史 3:13 可视化差异
I added labels for "Prolific" authors that made PRs to more than 20 repos in the last month. Some of the authors made PRs to 100 repos within 30 days 🤯 github.com/apache/arrow-rs/pulls