commit e07bc46c57656e1f652a59527f5e438ae414f940 Author: ilker Date: Wed Sep 16 01:00:03 2026 +0200 Faz 0: iskelet ve sim-engine Plan v0.2'nin Faz 0 çıktısı. Yazılan sıraya göre: - cobra tabanlı CLI iskeleti; server/agent/ctl komutları henüz stub. - internal/sim: OpenAI uyumlu sahte inference engine. Router'ı GPU olmadan gerçekçi koşullarda test edebilmek için gerçek bir engine'in router açısından önemli davranışlarını modeller: * continuous batching ve KV blok muhasebesi (head-of-line blocking dahil) * zincirlenmiş blok hash'leri + LRU ile otomatik prefix cache * batch büyüdükçe artan token gecikmesi * istemci koptuğunda isteğin düşürülüp blokların bırakılması * vLLM metrik adlarıyla /metrics (plan §7.6 engine kontratı) - Testler: prefix hash zincirlemesi, LRU tahliyesi, iptal propagasyonu, kuyruklama, SSE formatı, usage/cached_tokens, metriklerin gerçek Prometheus parser'ıyla okunabilirliği. - Gitea Actions CI: gofmt, vet, race detector. Modül yolu gitea.mynodes.xyz/ilker/llmgw; `make rename` ile değiştirilebilir. Co-Authored-By: Claude Opus 5 (1M context) diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..6721dd4 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,31 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version: "1.26" + cache: true + + - name: Biçim kontrolü + run: | + out=$(gofmt -l .) + if [ -n "$out" ]; then echo "gofmt gerekli:"; echo "$out"; exit 1; fi + + - name: Vet + run: go vet ./... + + - name: Test (race detector ile) + run: go test -race ./... + + - name: Build + run: go build ./... diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..31cc842 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +/bin/ +/dist/ +*.db +*.db-wal +*.db-shm +.DS_Store +coverage.out diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000..dcd2b00 --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,30 @@ +version: "2" + +linters: + default: standard + enable: + - bodyclose + - errorlint + - gocritic + - misspell + - noctx + - revive + - unconvert + - usetesting + + settings: + revive: + rules: + - name: exported + - name: package-comments + + exclusions: + rules: + # Testlerde okunabilirlik uğruna gevşetilenler. + - path: _test\.go + linters: [bodyclose, noctx, errorlint] + +formatters: + enable: + - gofmt + - goimports diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..b77063a --- /dev/null +++ b/Makefile @@ -0,0 +1,32 @@ +MODULE := gitea.mynodes.xyz/ilker/llmgw +VERSION ?= $(shell git describe --tags --always --dirty 2>/dev/null || echo dev) +LDFLAGS := -X '$(MODULE)/internal/version.Version=$(VERSION)' + +.PHONY: build test race lint fmt clean rename + +build: + go build -ldflags "$(LDFLAGS)" -o bin/llmgw ./cmd/llmgw + +test: + go test ./... + +race: + go test -race ./... + +lint: + golangci-lint run + +fmt: + go fmt ./... + go vet ./... + +clean: + rm -rf bin dist + +# Modül yolunu değiştirmek için: make rename NEW=git.ornek.com/ekip/llmgw +rename: + @test -n "$(NEW)" || (echo "kullanim: make rename NEW="; exit 1) + go mod edit -module=$(NEW) + grep -rl '$(MODULE)' --include='*.go' . | xargs sed -i '' 's|$(MODULE)|$(NEW)|g' + sed -i '' 's|^MODULE := .*|MODULE := $(NEW)|' Makefile + go build ./... diff --git a/README.md b/README.md new file mode 100644 index 0000000..82a561e --- /dev/null +++ b/README.md @@ -0,0 +1,69 @@ +# llmgw + +Çoklu InfiniBand fabric ve heterojen GPU için tek binary LLM gateway. + +Kendi GPU sunucularında çalışan LLM'leri (vLLM / SGLang) tek bir OpenAI uyumlu +uç noktadan sunar. Tasarım dokümanı: [llmgw-plan.md](llmgw-plan.md). + +## Durum + +| Faz | Kapsam | Durum | +|---|---|---| +| 0 | İskelet + sim-engine | ✅ | +| 1 | Statik router (filter/score/pick, streaming proxy) | ⬜ | +| 1.5 | Admission control (öncelik bantları) | ⬜ | +| 2 | Agent: GPU/IB keşfi, tek-node engine | ⬜ | +| 3 | Control plane: katalog, scheduler, reconciler | ⬜ | +| 4 | Multi-node & fabric farkındalığı | ⬜ | +| 5 | Tenancy, prefix affinity, bench | ⬜ | + +## Kurulum + +```sh +make build # bin/llmgw +make test # birim testleri +make race # yarış dedektörüyle +``` + +## sim-engine + +Gerçek bir inference engine'in **router açısından önemli** davranışlarını simüle +eder; GPU olmadan geliştirme ve test yapabilmek için. Modellediği şeyler: + +- continuous batching (sınırlı eşzamanlı dizi sayısı) +- KV cache blok muhasebesi; blok yetmediğinde kuyruğa alma (head-of-line blocking dahil) +- otomatik prefix cache (zincirlenmiş blok hash'leri + LRU) ve cache hit'in TTFT'ye etkisi +- batch büyüdükçe token başına gecikmenin artması +- istemci koptuğunda isteğin batch'ten düşürülüp blokların bırakılması +- vLLM metrik adlarıyla `/metrics` + +```sh +./bin/llmgw sim-engine --listen :8000 --model llama-3.3-70b + +curl -N localhost:8000/v1/chat/completions \ + -H 'content-type: application/json' \ + -d '{"messages":[{"role":"user","content":"merhaba"}],"max_tokens":8,"stream":true}' +``` + +Yük/gecikme davranışını ayarlamak için: `--max-num-seqs`, `--num-gpu-blocks`, +`--prefill-tps`, `--decode-tps`, `--batch-contention`, `--prefix-cache-blocks`. +Kaos testleri için: `--fail-rate`, deterministik koşu için `--seed`. + +### Engine metrik kontratı + +Router'ın yönlendirme yapabilmesi için bir engine'in vermesi **zorunlu** üç metrik +(plan §7.6). sim-engine bunları gerçek vLLM adlarıyla yayınlar: + +``` +vllm:num_requests_running +vllm:num_requests_waiting +vllm:kv_cache_usage_perc +``` + +## Modül yolu + +`gitea.mynodes.xyz/ilker/llmgw`. Değiştirmek için: + +```sh +make rename NEW=git.ornek.com/ekip/llmgw +``` diff --git a/cmd/llmgw/main.go b/cmd/llmgw/main.go new file mode 100644 index 0000000..19823ff --- /dev/null +++ b/cmd/llmgw/main.go @@ -0,0 +1,17 @@ +// Command llmgw, tek binary halinde router, control plane, agent ve yardımcı +// araçları barındırır. Hangi rolde çalışacağı alt komutla belirlenir. +package main + +import ( + "fmt" + "os" + + "gitea.mynodes.xyz/ilker/llmgw/internal/cli" +) + +func main() { + if err := cli.Root().Execute(); err != nil { + fmt.Fprintln(os.Stderr, "hata:", err) + os.Exit(1) + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..0f50395 --- /dev/null +++ b/go.mod @@ -0,0 +1,16 @@ +module gitea.mynodes.xyz/ilker/llmgw + +go 1.26.3 + +require ( + github.com/prometheus/common v0.71.0 + github.com/spf13/cobra v1.10.2 +) + +require ( + github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/spf13/pflag v1.0.9 // indirect + google.golang.org/protobuf v1.36.12 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..41032db --- /dev/null +++ b/go.sum @@ -0,0 +1,26 @@ +github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= +github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.71.0 h1:9KDAKb7Mj3HEVKyFCK6Dc/HIwlBzZIN2l7/lrHl3KK8= +github.com/prometheus/common v0.71.0/go.mod h1:CLJ5H8TEsGX8bl31BdMkfhIZ+QmZ9tBPPotUxUbfcmk= +github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= +github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= +github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= +github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/cli/root.go b/internal/cli/root.go new file mode 100644 index 0000000..1a7183e --- /dev/null +++ b/internal/cli/root.go @@ -0,0 +1,48 @@ +// Package cli, llmgw'nin komut satırı yüzeyini kurar. +package cli + +import ( + "log/slog" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/obs" + "gitea.mynodes.xyz/ilker/llmgw/internal/version" +) + +type globalFlags struct { + logLevel string + logFormat string +} + +var global globalFlags + +// Root, kök komutu ve tüm alt komutları döner. +func Root() *cobra.Command { + root := &cobra.Command{ + Use: "llmgw", + Short: "Çoklu IB fabric ve heterojen GPU için LLM gateway", + Long: "llmgw, kendi GPU sunucularında çalışan LLM'leri tek bir OpenAI uyumlu\n" + + "uç noktadan sunar. Tek binary, alt komutla farklı roller.", + Version: version.Version, + SilenceUsage: true, + SilenceErrors: true, + PersistentPreRun: func(_ *cobra.Command, _ []string) { + obs.Setup(obs.Options{Level: global.logLevel, Format: global.logFormat}) + }, + } + + pf := root.PersistentFlags() + pf.StringVar(&global.logLevel, "log-level", "info", "log seviyesi: debug|info|warn|error") + pf.StringVar(&global.logFormat, "log-format", "text", "log biçimi: text|json") + + root.AddCommand( + newServerCmd(), + newAgentCmd(), + newCtlCmd(), + newSimEngineCmd(), + ) + return root +} + +func logger() *slog.Logger { return slog.Default() } diff --git a/internal/cli/simengine.go b/internal/cli/simengine.go new file mode 100644 index 0000000..18b544b --- /dev/null +++ b/internal/cli/simengine.go @@ -0,0 +1,87 @@ +package cli + +import ( + "context" + "errors" + "net" + "net/http" + "os/signal" + "syscall" + "time" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/sim" +) + +func newSimEngineCmd() *cobra.Command { + cfg := sim.Default() + var listen string + + cmd := &cobra.Command{ + Use: "sim-engine", + Short: "OpenAI uyumlu sahte inference engine (test için)", + Long: "Gerçek bir engine'in router açısından önemli davranışlarını simüle eder:\n" + + "continuous batching, KV blok muhasebesi, prefix cache, vLLM formatında /metrics.\n" + + "Router'ı GPU olmadan test etmek için kullanılır.", + RunE: func(cmd *cobra.Command, _ []string) error { + log := logger() + eng := sim.New(cfg) + defer eng.Close() + + srv := &http.Server{ + Addr: listen, + Handler: sim.NewServer(eng, log), + ReadHeaderTimeout: 10 * time.Second, + } + + ln, err := net.Listen("tcp", listen) + if err != nil { + return err + } + log.Info("sim-engine dinliyor", + "addr", ln.Addr().String(), + "model", cfg.Model, + "max_num_seqs", cfg.MaxNumSeqs, + "num_gpu_blocks", cfg.NumGPUBlocks, + "prefix_cache_blocks", cfg.PrefixCacheBlocks, + ) + + ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + errCh := make(chan error, 1) + go func() { errCh <- srv.Serve(ln) }() + + select { + case err := <-errCh: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + case <-ctx.Done(): + log.Info("kapanıyor") + shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return srv.Shutdown(shutCtx) + } + }, + } + + f := cmd.Flags() + f.StringVar(&listen, "listen", ":8000", "dinlenecek adres") + f.StringVar(&cfg.Model, "model", cfg.Model, "sunulan model adı") + f.IntVar(&cfg.MaxNumSeqs, "max-num-seqs", cfg.MaxNumSeqs, "eşzamanlı dizi sayısı (batch kapasitesi)") + f.IntVar(&cfg.BlockSize, "block-size", cfg.BlockSize, "KV blok başına token") + f.IntVar(&cfg.NumGPUBlocks, "num-gpu-blocks", cfg.NumGPUBlocks, "toplam KV cache bloğu") + f.IntVar(&cfg.PrefixCacheBlocks, "prefix-cache-blocks", cfg.PrefixCacheBlocks, "prefix cache blok kapasitesi (0=kapalı)") + f.Float64Var(&cfg.PrefillTokensPerSec, "prefill-tps", cfg.PrefillTokensPerSec, "prefill hızı (token/sn, engine geneli)") + f.Float64Var(&cfg.DecodeTokensPerSec, "decode-tps", cfg.DecodeTokensPerSec, "tek dizi decode hızı (token/sn)") + f.Float64Var(&cfg.BatchContention, "batch-contention", cfg.BatchContention, "batch'teki her ek dizinin yavaşlatma katsayısı") + f.IntVar(&cfg.DefaultOutputTokens, "default-output-tokens", cfg.DefaultOutputTokens, "max_tokens verilmezse üretilecek token") + f.DurationVar(&cfg.Tick, "tick", cfg.Tick, "simülasyon adımı") + f.Float64Var(&cfg.FailRate, "fail-rate", 0, "isteklerin yapay hata oranı (kaos testi)") + f.Int64Var(&cfg.Seed, "seed", 0, "rastgelelik tohumu (0=zaman)") + + return cmd +} diff --git a/internal/cli/stubs.go b/internal/cli/stubs.go new file mode 100644 index 0000000..3613135 --- /dev/null +++ b/internal/cli/stubs.go @@ -0,0 +1,57 @@ +package cli + +import ( + "errors" + + "github.com/spf13/cobra" +) + +// errNotImplemented, henüz yazılmamış fazların komutları için. +var errNotImplemented = errors.New("bu komut henüz uygulanmadı (bkz. llmgw-plan.md §11)") + +func newServerCmd() *cobra.Command { + var configPath string + cmd := &cobra.Command{ + Use: "server", + Short: "Router + control plane (Faz 1-3)", + RunE: func(_ *cobra.Command, _ []string) error { + return errNotImplemented + }, + } + cmd.Flags().StringVar(&configPath, "config", "server.yaml", "konfigürasyon dosyası") + return cmd +} + +func newAgentCmd() *cobra.Command { + var configPath string + cmd := &cobra.Command{ + Use: "agent", + Short: "GPU node'unda çalışan agent (Faz 2)", + RunE: func(_ *cobra.Command, _ []string) error { + return errNotImplemented + }, + } + cmd.Flags().StringVar(&configPath, "config", "agent.yaml", "konfigürasyon dosyası") + return cmd +} + +func newCtlCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "ctl", + Short: "Yönetim CLI'ı (Faz 3)", + } + for _, sub := range []struct{ use, short string }{ + {"nodes", "node'ları listele"}, + {"models", "model kataloğunu göster"}, + {"replicas", "replikaları listele"}, + {"explain-placement", "bir modelin neden yerleştirilemediğini açıkla"}, + {"bench", "profil kapasitesini ölç (§10)"}, + } { + cmd.AddCommand(&cobra.Command{ + Use: sub.use, + Short: sub.short, + RunE: func(_ *cobra.Command, _ []string) error { return errNotImplemented }, + }) + } + return cmd +} diff --git a/internal/obs/log.go b/internal/obs/log.go new file mode 100644 index 0000000..c4f6952 --- /dev/null +++ b/internal/obs/log.go @@ -0,0 +1,41 @@ +// Package obs, log ve metrik kurulumunu tek yerden yapar. +package obs + +import ( + "log/slog" + "os" + "strings" +) + +// Options, log kurulumunu belirler. +type Options struct { + Level string // debug, info, warn, error + Format string // text, json +} + +// Setup, slog varsayılan logger'ını kurar ve onu döner. +func Setup(o Options) *slog.Logger { + var lvl slog.Level + switch strings.ToLower(o.Level) { + case "debug": + lvl = slog.LevelDebug + case "warn": + lvl = slog.LevelWarn + case "error": + lvl = slog.LevelError + default: + lvl = slog.LevelInfo + } + + hopts := &slog.HandlerOptions{Level: lvl} + var h slog.Handler + if strings.EqualFold(o.Format, "text") { + h = slog.NewTextHandler(os.Stderr, hopts) + } else { + h = slog.NewJSONHandler(os.Stderr, hopts) + } + + l := slog.New(h) + slog.SetDefault(l) + return l +} diff --git a/internal/sim/cache.go b/internal/sim/cache.go new file mode 100644 index 0000000..9c61a0c --- /dev/null +++ b/internal/sim/cache.go @@ -0,0 +1,128 @@ +package sim + +import ( + "container/list" + "encoding/binary" + "hash/fnv" +) + +// BlockHash, KV cache'in muhasebe birimi olan bir bloğun kimliği. +type BlockHash uint64 + +// BlockHashes, token dizisi için zincirlenmiş blok hash'leri üretir. +// +// Zincirleme (h_i = hash(h_{i-1}, block_i)) vLLM'in yaptığının aynısıdır ve +// şart: blok kimliği sadece kendi içeriğini değil, kendisinden önce gelen tüm +// blokları da temsil etmeli. Aksi halde "aynı cümleyi farklı bağlamda içeren" +// iki istek yanlışlıkla cache hit sayılır. +// +// Son eksik blok yok sayılır — yarım blok cache'lenmez. +func BlockHashes(tokens []Token, blockSize int) []BlockHash { + if blockSize <= 0 || len(tokens) < blockSize { + return nil + } + n := len(tokens) / blockSize + out := make([]BlockHash, 0, n) + var prev BlockHash + buf := make([]byte, 8) + for i := range n { + h := fnv.New64a() + binary.LittleEndian.PutUint64(buf, uint64(prev)) + _, _ = h.Write(buf) + for _, t := range tokens[i*blockSize : (i+1)*blockSize] { + binary.LittleEndian.PutUint32(buf[:4], uint32(t)) + _, _ = h.Write(buf[:4]) + } + prev = BlockHash(h.Sum64()) + out = append(out, prev) + } + return out +} + +// PrefixCache, engine'in otomatik prefix cache'ini modelleyen blok bazlı LRU. +// +// Gerçek engine'de bu bir radix tree'dir; ama yönlendirme kalitesini ölçmek için +// önemli olan davranış "hangi blokların hâlâ cache'te olduğu"dur ve onu LRU +// doğru modeller. +type PrefixCache struct { + capacity int // blok cinsinden + ll *list.List + m map[BlockHash]*list.Element + + queries int64 + hits int64 +} + +// NewPrefixCache, verilen blok kapasitesinde bir cache kurar. +// capacity <= 0 ise cache devre dışıdır (her zaman miss). +func NewPrefixCache(capacity int) *PrefixCache { + return &PrefixCache{ + capacity: capacity, + ll: list.New(), + m: make(map[BlockHash]*list.Element), + } +} + +// Enabled, cache'in açık olup olmadığını söyler. +func (c *PrefixCache) Enabled() bool { return c != nil && c.capacity > 0 } + +// Lookup, hashes dizisinin cache'te bulunan **kesintisiz ön ek** uzunluğunu +// blok cinsinden döner ve bulunan blokları LRU'da tazeler. +// +// Kesintisiz olması önemli: ortadaki bir blok düşmüşse ondan sonrası da +// yeniden hesaplanmak zorundadır, çünkü attention önceki tüm bloklara bağlıdır. +func (c *PrefixCache) Lookup(hashes []BlockHash) int { + if !c.Enabled() { + return 0 + } + c.queries += int64(len(hashes)) + n := 0 + for _, h := range hashes { + el, ok := c.m[h] + if !ok { + break + } + c.ll.MoveToFront(el) + n++ + } + c.hits += int64(n) + return n +} + +// Insert, blokları cache'e yazar (LRU'nun başına) ve kapasiteyi aşarsa en eskileri atar. +func (c *PrefixCache) Insert(hashes []BlockHash) { + if !c.Enabled() { + return + } + for _, h := range hashes { + if el, ok := c.m[h]; ok { + c.ll.MoveToFront(el) + continue + } + c.m[h] = c.ll.PushFront(h) + } + for c.ll.Len() > c.capacity { + el := c.ll.Back() + if el == nil { + break + } + c.ll.Remove(el) + delete(c.m, el.Value.(BlockHash)) + } +} + +// Stats, cache sorgu ve isabet sayılarını döner (blok cinsinden). +func (c *PrefixCache) Stats() (queries, hits int64) { + if c == nil { + return 0, 0 + } + return c.queries, c.hits +} + +// Len, cache'teki blok sayısı. +func (c *PrefixCache) Len() int { + if !c.Enabled() { + return 0 + } + return c.ll.Len() +} diff --git a/internal/sim/cache_test.go b/internal/sim/cache_test.go new file mode 100644 index 0000000..d27320f --- /dev/null +++ b/internal/sim/cache_test.go @@ -0,0 +1,77 @@ +package sim + +import "testing" + +func TestBlockHashesZincirleme(t *testing.T) { + // Aynı içerik farklı ön ekten sonra geliyorsa blok hash'i farklı olmalı. + a := BlockHashes(Tokenize("AAAAAAAABBBBBBBB"), 2) + b := BlockHashes(Tokenize("CCCCCCCCBBBBBBBB"), 2) + if len(a) != 2 || len(b) != 2 { + t.Fatalf("beklenen 2 blok, alınan a=%d b=%d", len(a), len(b)) + } + if a[1] == b[1] { + t.Error("farklı ön ekten sonra gelen aynı blok aynı hash'i verdi; zincirleme çalışmıyor") + } +} + +func TestBlockHashesOnEkKararli(t *testing.T) { + // Blok sınırına hizalı bir ön ek, uzun metinde aynı blok hash'lerini vermeli. + // Hizalama şart: son yarım token farklı olursa onu içeren blok da farklı + // çıkar — gerçek tokenizer'larda da olan sınır etkisi, bu yüzden prefix + // cache hit'i her zaman tam blok cinsinden ölçülür. + const aligned = "merhaba " // 8 bayt = 2 token = blockSize 2 ile 1 tam blok + short := BlockHashes(Tokenize(aligned), 2) + long := BlockHashes(Tokenize(aligned+"dunya nasilsin"), 2) + if len(short) != 1 { + t.Fatalf("beklenen 1 blok, alınan %d", len(short)) + } + if short[0] != long[0] { + t.Fatal("hizalı ön ek blok hash'i değişti; ön ek kararlı değil") + } +} + +func TestBlockHashesYarimBlokAtilir(t *testing.T) { + // 3 token, blockSize 2 -> 1 tam blok + if got := len(BlockHashes(make([]Token, 3), 2)); got != 1 { + t.Errorf("beklenen 1 blok, alınan %d", got) + } + if got := len(BlockHashes(make([]Token, 1), 2)); got != 0 { + t.Errorf("tam blok yokken blok üretildi: %d", got) + } +} + +func TestPrefixCacheKesintisizOnEk(t *testing.T) { + c := NewPrefixCache(10) + h := []BlockHash{1, 2, 3, 4} + c.Insert(h) + + if n := c.Lookup(h); n != 4 { + t.Fatalf("tam eşleşme beklendi, alınan %d", n) + } + // Ortası eksik: ilk blok hit, sonrası miss olmalı. + if n := c.Lookup([]BlockHash{1, 99, 3}); n != 1 { + t.Errorf("kesintisiz ön ek 1 olmalıydı, alınan %d", n) + } +} + +func TestPrefixCacheLRUTahliye(t *testing.T) { + c := NewPrefixCache(2) + c.Insert([]BlockHash{1, 2}) + c.Lookup([]BlockHash{1}) // 1'i tazele + c.Insert([]BlockHash{3}) // kapasite aşıldı, en eski (2) düşmeli + + if n := c.Lookup([]BlockHash{1}); n != 1 { + t.Error("taze blok tahliye edildi") + } + if n := c.Lookup([]BlockHash{2}); n != 0 { + t.Error("en eski blok tahliye edilmeliydi") + } +} + +func TestPrefixCacheKapali(t *testing.T) { + c := NewPrefixCache(0) + c.Insert([]BlockHash{1, 2}) + if n := c.Lookup([]BlockHash{1}); n != 0 { + t.Errorf("kapalı cache hit verdi: %d", n) + } +} diff --git a/internal/sim/engine.go b/internal/sim/engine.go new file mode 100644 index 0000000..48ace20 --- /dev/null +++ b/internal/sim/engine.go @@ -0,0 +1,510 @@ +// Package sim, OpenAI uyumlu sahte bir inference engine'i barındırır. +// +// Amacı router'ı GPU olmadan gerçekçi koşullarda test edebilmek. Gerçek bir +// engine'in router açısından önemli olan davranışlarını modeller: +// - continuous batching (sınırlı eşzamanlı dizi sayısı) +// - KV cache blok muhasebesi ve blok yetmediğinde kuyruğa alma +// - otomatik prefix cache (blok bazlı LRU) ve cache hit'in TTFT'ye etkisi +// - batch büyüdükçe token başına gecikmenin artması +// - vLLM formatında /metrics +package sim + +import ( + "context" + "errors" + "fmt" + "math/rand" + "sync" + "sync/atomic" + "time" +) + +// Config, simüle edilen engine'in davranışını belirler. +type Config struct { + Model string + + // MaxNumSeqs, aynı anda çalışabilecek dizi sayısı (continuous batch kapasitesi). + MaxNumSeqs int + // BlockSize, bir KV cache bloğundaki token sayısı. + BlockSize int + // NumGPUBlocks, toplam KV cache bloğu. + NumGPUBlocks int + // PrefixCacheBlocks, prefix cache'in tutacağı blok sayısı. 0 = kapalı. + PrefixCacheBlocks int + + // PrefillTokensPerSec, engine genelinde paylaşılan prefill hızı. + PrefillTokensPerSec float64 + // DecodeTokensPerSec, tek dizi çalışırken token üretim hızı. + DecodeTokensPerSec float64 + // BatchContention, batch'teki her ek dizinin decode hızını ne kadar + // yavaşlattığı. Etkin hız = DecodeTokensPerSec / (1 + BatchContention*(n-1)). + BatchContention float64 + + // DefaultOutputTokens, istek max_tokens vermediğinde üretilecek token sayısı. + DefaultOutputTokens int + // Tick, simülasyon adımı. Küçük olması çözünürlüğü artırır. + Tick time.Duration + + // FailRate, isteklerin hangi oranda 500 ile reddedileceği (kaos testi için). + FailRate float64 + // Seed, deterministik testler için. + Seed int64 +} + +// Default, makul varsayılanlarla bir Config döner. +// Kabaca 8xH100 üstünde 70B bir modelin davranışına benzer ölçekler. +func Default() Config { + return Config{ + Model: "sim-model", + MaxNumSeqs: 64, + BlockSize: 16, + NumGPUBlocks: 4096, + PrefixCacheBlocks: 8192, + PrefillTokensPerSec: 12000, + DecodeTokensPerSec: 60, + BatchContention: 0.012, + DefaultOutputTokens: 128, + Tick: 5 * time.Millisecond, + } +} + +func (c *Config) normalize() { + if c.Model == "" { + c.Model = "sim-model" + } + if c.MaxNumSeqs <= 0 { + c.MaxNumSeqs = 64 + } + if c.BlockSize <= 0 { + c.BlockSize = 16 + } + if c.NumGPUBlocks <= 0 { + c.NumGPUBlocks = 4096 + } + if c.PrefillTokensPerSec <= 0 { + c.PrefillTokensPerSec = 12000 + } + if c.DecodeTokensPerSec <= 0 { + c.DecodeTokensPerSec = 60 + } + if c.DefaultOutputTokens <= 0 { + c.DefaultOutputTokens = 128 + } + if c.Tick <= 0 { + c.Tick = 5 * time.Millisecond + } +} + +// EventKind, akıştaki olay türü. +type EventKind int + +const ( + // EventToken, üretilen bir token. + EventToken EventKind = iota + // EventDone, üretimin normal bitişi. + EventDone + // EventError, üretim sırasında oluşan hata. + EventError +) + +// Event, engine'in istemciye ilettiği tek bir olay. +type Event struct { + Kind EventKind + Text string + Err error + // Result yalnızca EventDone'da doludur; prompt/cache/completion sayılarını taşır. + Result *Result +} + +// Prompt, engine'e verilen istek. +type Prompt struct { + Text string + MaxTokens int +} + +// Result, istek bittiğinde okunabilen özet (Done olayından sonra geçerli). +type Result struct { + PromptTokens int + CachedTokens int + CompletionTokens int +} + +var ( + // ErrTooLong, isteğin KV cache'e hiç sığmadığını belirtir. + ErrTooLong = errors.New("istek KV cache kapasitesine sığmıyor") + // ErrInjected, FailRate ile üretilen yapay hata. + ErrInjected = errors.New("yapay hata (sim fail_rate)") + // ErrShutdown, engine kapanırken bekleyen istekler için. + ErrShutdown = errors.New("engine kapanıyor") +) + +type req struct { + id string + ctx context.Context + prompt Prompt + + promptTokens int + cachedTokens int + outputTokens int + hashes []BlockHash + + blocks int + prefillLeft float64 + decodeLeft int + decodeAccum float64 + emitted int + + arrived time.Time + firstAt time.Time + lastTokAt time.Time + + out chan Event + pending []Event + finished bool + res *Result +} + +// flush, biriken olayları bloke olmadan kanala aktarır. +// Bloke olmamak şart: tick döngüsü kilidi tutarken yavaş bir istemci +// tüm engine'i durdurmamalı. +func (r *req) flush() { + for len(r.pending) > 0 { + select { + case r.out <- r.pending[0]: + r.pending = r.pending[1:] + default: + return + } + } +} + +func (r *req) emit(ev Event) { + r.pending = append(r.pending, ev) + r.flush() +} + +// Engine, simüle edilen inference engine. +type Engine struct { + cfg Config + + mu sync.Mutex + waiting []*req + running []*req + draining []*req + freeBlocks int + cache *PrefixCache + rng *rand.Rand + seq atomic.Uint64 + + m *metrics + + stopOnce sync.Once + stop chan struct{} + done chan struct{} +} + +// New, verilen konfigürasyonla bir engine kurar ve simülasyon döngüsünü başlatır. +// Kapatmak için Close çağrılmalı. +func New(cfg Config) *Engine { + cfg.normalize() + seed := cfg.Seed + if seed == 0 { + seed = time.Now().UnixNano() + } + e := &Engine{ + cfg: cfg, + freeBlocks: cfg.NumGPUBlocks, + cache: NewPrefixCache(cfg.PrefixCacheBlocks), + rng: rand.New(rand.NewSource(seed)), //nolint:gosec // simülasyon + m: newMetrics(), + stop: make(chan struct{}), + done: make(chan struct{}), + } + go e.run() + return e +} + +// Config, engine'in aktif konfigürasyonunu döner. +func (e *Engine) Config() Config { return e.cfg } + +// Close, simülasyon döngüsünü durdurur ve bekleyen istekleri hataya düşürür. +func (e *Engine) Close() { + e.stopOnce.Do(func() { + close(e.stop) + <-e.done + }) +} + +// Submit, isteği kuyruğa alır ve olay kanalını döner. +// Kanal, üretim bitince veya hata oluşunca kapanır. +func (e *Engine) Submit(ctx context.Context, p Prompt) (<-chan Event, error) { + if e.cfg.FailRate > 0 { + e.mu.Lock() + fail := e.rng.Float64() < e.cfg.FailRate + e.mu.Unlock() + if fail { + e.m.failed.Add(1) + return nil, ErrInjected + } + } + + tokens := Tokenize(p.Text) + out := p.MaxTokens + if out <= 0 { + out = e.cfg.DefaultOutputTokens + } + + r := &req{ + id: fmt.Sprintf("sim-%d", e.seq.Add(1)), + ctx: ctx, + prompt: p, + promptTokens: len(tokens), + outputTokens: out, + hashes: BlockHashes(tokens, e.cfg.BlockSize), + arrived: time.Now(), + out: make(chan Event, 64), + } + + if e.blocksFor(r.promptTokens+r.outputTokens) > e.cfg.NumGPUBlocks { + e.m.failed.Add(1) + return nil, ErrTooLong + } + + e.mu.Lock() + select { + case <-e.stop: + e.mu.Unlock() + return nil, ErrShutdown + default: + } + e.waiting = append(e.waiting, r) + e.mu.Unlock() + + e.m.received.Add(1) + return r.out, nil +} + +func (e *Engine) blocksFor(tokens int) int { + return (tokens + e.cfg.BlockSize - 1) / e.cfg.BlockSize +} + +func (e *Engine) run() { + defer close(e.done) + t := time.NewTicker(e.cfg.Tick) + defer t.Stop() + dt := e.cfg.Tick.Seconds() + + for { + select { + case <-e.stop: + e.shutdown() + return + case <-t.C: + e.mu.Lock() + e.reap() + e.schedule() + e.advance(dt) + e.flushAll() + e.mu.Unlock() + } + } +} + +// reap, istemcisi vazgeçmiş istekleri hem kuyruktan hem batch'ten düşürür. +// Router'ın iptal propagasyonunu test edebilmek için bu davranış şart: +// istemci koptuğunda GPU boşa çalışmamalı. +func (e *Engine) reap() { + e.waiting = filterInPlace(e.waiting, func(r *req) bool { + if r.ctx.Err() == nil { + return true + } + e.m.cancelled.Add(1) + close(r.out) + return false + }) + e.running = filterInPlace(e.running, func(r *req) bool { + if r.ctx.Err() == nil { + return true + } + e.freeBlocks += r.blocks + e.m.cancelled.Add(1) + close(r.out) + return false + }) +} + +func (e *Engine) schedule() { + for len(e.waiting) > 0 && len(e.running) < e.cfg.MaxNumSeqs { + r := e.waiting[0] + need := e.blocksFor(r.promptTokens + r.outputTokens) + if need > e.freeBlocks { + // Head-of-line blocking: gerçek engine de böyle davranır. + break + } + + hitBlocks := e.cache.Lookup(r.hashes) + r.cachedTokens = hitBlocks * e.cfg.BlockSize + r.blocks = need + r.prefillLeft = float64(r.promptTokens - r.cachedTokens) + if r.prefillLeft < 0 { + r.prefillLeft = 0 + } + r.decodeLeft = r.outputTokens + + e.freeBlocks -= need + e.running = append(e.running, r) + e.waiting = e.waiting[1:] + } +} + +func (e *Engine) advance(dt float64) { + if len(e.running) == 0 { + return + } + + // Prefill bütçesi engine genelinde paylaşılır ve FIFO dağıtılır. + budget := e.cfg.PrefillTokensPerSec * dt + for _, r := range e.running { + if budget <= 0 { + break + } + if r.prefillLeft <= 0 { + continue + } + used := min(budget, r.prefillLeft) + r.prefillLeft -= used + budget -= used + } + + // Decode hızı batch büyüdükçe düşer. + n := 0 + for _, r := range e.running { + if r.prefillLeft <= 0 { + n++ + } + } + if n == 0 { + return + } + rate := e.cfg.DecodeTokensPerSec / (1 + e.cfg.BatchContention*float64(n-1)) + + now := time.Now() + for _, r := range e.running { + if r.prefillLeft > 0 || r.finished { + continue + } + r.decodeAccum += rate * dt + for r.decodeAccum >= 1 && r.decodeLeft > 0 { + r.decodeAccum-- + r.decodeLeft-- + r.emitted++ + + if r.firstAt.IsZero() { + r.firstAt = now + e.m.observeTTFT(now.Sub(r.arrived)) + } else { + e.m.observeTPOT(now.Sub(r.lastTokAt)) + } + r.lastTokAt = now + r.emit(Event{Kind: EventToken, Text: tokenText(r.emitted)}) + } + if r.decodeLeft == 0 { + e.finish(r) + } + } + + e.running = filterInPlace(e.running, func(r *req) bool { return !r.finished }) +} + +func (e *Engine) finish(r *req) { + r.finished = true + e.freeBlocks += r.blocks + + // Üretilen çıktı da cache'e girer: bir sonraki turda konuşma geçmişi + // olarak geri geldiğinde hit vermeli. Çok turlu sohbeti modellemenin + // tek yolu bu. + full := Tokenize(r.prompt.Text + generatedText(r.emitted)) + e.cache.Insert(BlockHashes(full, e.cfg.BlockSize)) + + r.res = &Result{ + PromptTokens: r.promptTokens, + CachedTokens: r.cachedTokens, + CompletionTokens: r.emitted, + } + e.m.finished.Add(1) + e.m.promptTokens.Add(int64(r.promptTokens)) + e.m.genTokens.Add(int64(r.emitted)) + + r.emit(Event{Kind: EventDone, Result: r.res}) + e.draining = append(e.draining, r) +} + +func (e *Engine) flushAll() { + for _, r := range e.running { + r.flush() + } + e.draining = filterInPlace(e.draining, func(r *req) bool { + r.flush() + if len(r.pending) == 0 { + close(r.out) + return false + } + return true + }) +} + +func (e *Engine) shutdown() { + e.mu.Lock() + defer e.mu.Unlock() + for _, r := range append(e.waiting, e.running...) { + select { + case r.out <- Event{Kind: EventError, Err: ErrShutdown}: + default: + } + close(r.out) + } + for _, r := range e.draining { + close(r.out) + } + e.waiting, e.running, e.draining = nil, nil, nil +} + +// Stats, router'ın scrape edeceği anlık durumu döner. +func (e *Engine) Stats() Stats { + e.mu.Lock() + defer e.mu.Unlock() + used := e.cfg.NumGPUBlocks - e.freeBlocks + q, h := e.cache.Stats() + return Stats{ + Running: len(e.running), + Waiting: len(e.waiting), + KVCacheUsage: float64(used) / float64(e.cfg.NumGPUBlocks), + BlockSize: e.cfg.BlockSize, + NumGPUBlocks: e.cfg.NumGPUBlocks, + PrefixQueries: q, + PrefixHits: h, + CachedBlocks: e.cache.Len(), + } +} + +// Stats, engine'in anlık durumu. +type Stats struct { + Running int + Waiting int + KVCacheUsage float64 + BlockSize int + NumGPUBlocks int + PrefixQueries int64 + PrefixHits int64 + CachedBlocks int +} + +func filterInPlace(s []*req, keep func(*req) bool) []*req { + out := s[:0] + for _, r := range s { + if keep(r) { + out = append(out, r) + } + } + return out +} diff --git a/internal/sim/engine_test.go b/internal/sim/engine_test.go new file mode 100644 index 0000000..4a62051 --- /dev/null +++ b/internal/sim/engine_test.go @@ -0,0 +1,247 @@ +package sim + +import ( + "context" + "strings" + "testing" + "time" +) + +// testConfig, testlerin hızlı ve deterministik koşması için hızlandırılmış ayarlar. +func testConfig() Config { + c := Default() + c.Tick = time.Millisecond + c.PrefillTokensPerSec = 1e6 + c.DecodeTokensPerSec = 5000 + c.BatchContention = 0 + c.Seed = 42 + return c +} + +// drain, kanaldaki tüm token'ları toplar ve bitişi bekler. +func drain(t *testing.T, ch <-chan Event) (text string, done bool) { + t.Helper() + var b strings.Builder + deadline := time.After(10 * time.Second) + for { + select { + case ev, open := <-ch: + if !open { + return b.String(), done + } + switch ev.Kind { + case EventToken: + b.WriteString(ev.Text) + case EventDone: + done = true + case EventError: + t.Fatalf("beklenmeyen hata: %v", ev.Err) + } + case <-deadline: + t.Fatal("zaman aşımı: engine token üretmedi") + } + } +} + +func TestEngineTemelUretim(t *testing.T) { + e := New(testConfig()) + defer e.Close() + + ch, err := e.Submit(t.Context(), Prompt{Text: "merhaba dunya", MaxTokens: 10}) + if err != nil { + t.Fatal(err) + } + text, done := drain(t, ch) + if !done { + t.Error("EventDone gelmedi") + } + if got := len(strings.Fields(text)); got != 10 { + t.Errorf("beklenen 10 token, alınan %d (%q)", got, text) + } +} + +func TestEngineIptalBloklariSerbestBirakir(t *testing.T) { + // Router'ın iptal propagasyonunu test edebilmesi için kritik davranış: + // istemci koptuğunda engine isteği düşürmeli ve KV bloklarını bırakmalı. + cfg := testConfig() + cfg.DecodeTokensPerSec = 20 // uzun sürsün ki iptal ederken çalışıyor olsun + e := New(cfg) + defer e.Close() + + ctx, cancel := context.WithCancel(context.Background()) + ch, err := e.Submit(ctx, Prompt{Text: strings.Repeat("x", 4000), MaxTokens: 1000}) + if err != nil { + t.Fatal(err) + } + + waitFor(t, time.Second, func() bool { return e.Stats().Running == 1 }) + if usage := e.Stats().KVCacheUsage; usage <= 0 { + t.Fatal("çalışan istek KV bloğu ayırmadı") + } + + cancel() + + waitFor(t, 2*time.Second, func() bool { + s := e.Stats() + return s.Running == 0 && s.KVCacheUsage == 0 + }) + + // Kanal kapanmalı ki istemci tarafı bloke kalmasın. + select { + case _, open := <-ch: + if open { + // token kalmış olabilir; kapanana kadar tüket + for range ch { //nolint:revive // kanalı boşalt + } + } + case <-time.After(2 * time.Second): + t.Fatal("iptal sonrası kanal kapanmadı") + } +} + +func TestEngineBatchDolunkaKuyrugaAlir(t *testing.T) { + cfg := testConfig() + cfg.MaxNumSeqs = 2 + cfg.DecodeTokensPerSec = 20 + e := New(cfg) + defer e.Close() + + for range 5 { + if _, err := e.Submit(t.Context(), Prompt{Text: "abc", MaxTokens: 100}); err != nil { + t.Fatal(err) + } + } + + waitFor(t, time.Second, func() bool { return e.Stats().Running == 2 }) + s := e.Stats() + if s.Running != 2 { + t.Errorf("running %d, MaxNumSeqs=2 aşıldı", s.Running) + } + if s.Waiting != 3 { + t.Errorf("waiting %d, beklenen 3", s.Waiting) + } +} + +func TestEngineKVBlogunaSigmayanIstekReddedilir(t *testing.T) { + cfg := testConfig() + cfg.NumGPUBlocks = 4 + cfg.BlockSize = 16 // toplam 64 token kapasite + e := New(cfg) + defer e.Close() + + _, err := e.Submit(t.Context(), Prompt{Text: strings.Repeat("x", 10000), MaxTokens: 100}) + if err != ErrTooLong { + t.Errorf("ErrTooLong beklendi, alınan %v", err) + } +} + +func TestEnginePrefixCacheIkinciIstegiHizlandirir(t *testing.T) { + cfg := testConfig() + cfg.PrefillTokensPerSec = 2000 // prefill maliyeti ölçülebilir olsun + cfg.DecodeTokensPerSec = 5000 + e := New(cfg) + defer e.Close() + + prompt := Prompt{Text: strings.Repeat("ortak sistem promptu. ", 200), MaxTokens: 5} + + first := timeToFirstToken(t, e, prompt) + // İlk istek bitip cache'e yazıldıktan sonra aynısını tekrar gönder. + second := timeToFirstToken(t, e, prompt) + + if second >= first { + t.Errorf("prefix cache TTFT'yi düşürmedi: ilk=%v ikinci=%v", first, second) + } + + s := e.Stats() + if s.PrefixHits == 0 { + t.Error("prefix cache hit sayacı artmadı") + } +} + +func TestEnginePrefixCacheKapaliykenHitYok(t *testing.T) { + cfg := testConfig() + cfg.PrefixCacheBlocks = 0 + e := New(cfg) + defer e.Close() + + p := Prompt{Text: strings.Repeat("ayni prompt. ", 50), MaxTokens: 3} + for range 3 { + ch, err := e.Submit(t.Context(), p) + if err != nil { + t.Fatal(err) + } + drain(t, ch) + } + if h := e.Stats().PrefixHits; h != 0 { + t.Errorf("cache kapalıyken %d hit raporlandı", h) + } +} + +func TestEngineKapanisBekleyenleriBosaltir(t *testing.T) { + cfg := testConfig() + cfg.DecodeTokensPerSec = 1 + e := New(cfg) + + ch, err := e.Submit(t.Context(), Prompt{Text: "abc", MaxTokens: 1000}) + if err != nil { + t.Fatal(err) + } + waitFor(t, time.Second, func() bool { return e.Stats().Running == 1 }) + e.Close() + + deadline := time.After(2 * time.Second) + for { + select { + case _, open := <-ch: + if !open { + return + } + case <-deadline: + t.Fatal("kapanışta kanal kapanmadı") + } + } +} + +func TestEngineYapayHata(t *testing.T) { + cfg := testConfig() + cfg.FailRate = 1.0 + e := New(cfg) + defer e.Close() + + if _, err := e.Submit(t.Context(), Prompt{Text: "abc"}); err != ErrInjected { + t.Errorf("ErrInjected beklendi, alınan %v", err) + } +} + +// --- yardımcılar --- + +func timeToFirstToken(t *testing.T, e *Engine, p Prompt) time.Duration { + t.Helper() + start := time.Now() + ch, err := e.Submit(t.Context(), p) + if err != nil { + t.Fatal(err) + } + var ttft time.Duration + for ev := range ch { + if ev.Kind == EventToken && ttft == 0 { + ttft = time.Since(start) + } + } + if ttft == 0 { + t.Fatal("hiç token üretilmedi") + } + return ttft +} + +func waitFor(t *testing.T, limit time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(limit) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("koşul %v içinde sağlanmadı", limit) +} diff --git a/internal/sim/metrics.go b/internal/sim/metrics.go new file mode 100644 index 0000000..1132b0f --- /dev/null +++ b/internal/sim/metrics.go @@ -0,0 +1,142 @@ +package sim + +import ( + "fmt" + "io" + "sync" + "sync/atomic" + "time" +) + +// ttftBuckets ve tpotBuckets, vLLM'in kullandıklarına yakın saniye sınırları. +var ( + ttftBuckets = []float64{0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, 20, 40, 80} + tpotBuckets = []float64{0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1, 2.5} +) + +// histogram, Prometheus uyumlu basit bir kümülatif histogram. +type histogram struct { + bounds []float64 + counts []uint64 + sum float64 + count uint64 +} + +func newHistogram(bounds []float64) *histogram { + return &histogram{bounds: bounds, counts: make([]uint64, len(bounds))} +} + +func (h *histogram) observe(v float64) { + h.sum += v + h.count++ + for i, b := range h.bounds { + if v <= b { + h.counts[i]++ + } + } +} + +// write, histogramı Prometheus text formatında yazar (kümülatif le bucket'ları). +func (h *histogram) write(w io.Writer, name, labels string) { + var cum uint64 + for i, b := range h.bounds { + cum = h.counts[i] + fmt.Fprintf(w, "%s_bucket{%s,le=\"%g\"} %d\n", name, labels, b, cum) + } + fmt.Fprintf(w, "%s_bucket{%s,le=\"+Inf\"} %d\n", name, labels, h.count) + fmt.Fprintf(w, "%s_sum{%s} %g\n", name, labels, h.sum) + fmt.Fprintf(w, "%s_count{%s} %d\n", name, labels, h.count) +} + +type metrics struct { + received atomic.Int64 + finished atomic.Int64 + failed atomic.Int64 + cancelled atomic.Int64 + + promptTokens atomic.Int64 + genTokens atomic.Int64 + + mu sync.Mutex + ttft *histogram + tpot *histogram +} + +func newMetrics() *metrics { + return &metrics{ + ttft: newHistogram(ttftBuckets), + tpot: newHistogram(tpotBuckets), + } +} + +func (m *metrics) observeTTFT(d time.Duration) { + m.mu.Lock() + m.ttft.observe(d.Seconds()) + m.mu.Unlock() +} + +func (m *metrics) observeTPOT(d time.Duration) { + m.mu.Lock() + m.tpot.observe(d.Seconds()) + m.mu.Unlock() +} + +// WriteMetrics, engine durumunu vLLM'in metrik adlarıyla Prometheus text +// formatında yazar. +// +// Metrik adları §7.6'daki engine kontratıdır; router bunlara göre parse eder. +// Gerçek vLLM sürümleriyle aynı adları kullanmak şart — sim-engine'in varlık +// sebebi router'ın gerçek engine'e karşı davranışını taklit etmek. +func (e *Engine) WriteMetrics(w io.Writer) { + s := e.Stats() + lbl := fmt.Sprintf("model_name=%q", e.cfg.Model) + + fmt.Fprintf(w, "# HELP vllm:num_requests_running Number of requests currently running on GPU.\n") + fmt.Fprintf(w, "# TYPE vllm:num_requests_running gauge\n") + fmt.Fprintf(w, "vllm:num_requests_running{%s} %d\n", lbl, s.Running) + + fmt.Fprintf(w, "# HELP vllm:num_requests_waiting Number of requests waiting to be processed.\n") + fmt.Fprintf(w, "# TYPE vllm:num_requests_waiting gauge\n") + fmt.Fprintf(w, "vllm:num_requests_waiting{%s} %d\n", lbl, s.Waiting) + + fmt.Fprintf(w, "# HELP vllm:kv_cache_usage_perc KV-cache usage. 1 means 100 percent usage.\n") + fmt.Fprintf(w, "# TYPE vllm:kv_cache_usage_perc gauge\n") + fmt.Fprintf(w, "vllm:kv_cache_usage_perc{%s} %g\n", lbl, s.KVCacheUsage) + + fmt.Fprintf(w, "# HELP vllm:cache_config_info Information of the LLMEngine CacheConfig.\n") + fmt.Fprintf(w, "# TYPE vllm:cache_config_info gauge\n") + fmt.Fprintf(w, "vllm:cache_config_info{block_size=\"%d\",num_gpu_blocks=\"%d\"} 1\n", s.BlockSize, s.NumGPUBlocks) + + fmt.Fprintf(w, "# HELP vllm:prefix_cache_queries_total Prefix cache queries, in terms of number of queried blocks.\n") + fmt.Fprintf(w, "# TYPE vllm:prefix_cache_queries_total counter\n") + fmt.Fprintf(w, "vllm:prefix_cache_queries_total{%s} %d\n", lbl, s.PrefixQueries) + + fmt.Fprintf(w, "# HELP vllm:prefix_cache_hits_total Prefix cache hits, in terms of number of cached blocks.\n") + fmt.Fprintf(w, "# TYPE vllm:prefix_cache_hits_total counter\n") + fmt.Fprintf(w, "vllm:prefix_cache_hits_total{%s} %d\n", lbl, s.PrefixHits) + + fmt.Fprintf(w, "# HELP vllm:prompt_tokens_total Number of prefill tokens processed.\n") + fmt.Fprintf(w, "# TYPE vllm:prompt_tokens_total counter\n") + fmt.Fprintf(w, "vllm:prompt_tokens_total{%s} %d\n", lbl, e.m.promptTokens.Load()) + + fmt.Fprintf(w, "# HELP vllm:generation_tokens_total Number of generation tokens processed.\n") + fmt.Fprintf(w, "# TYPE vllm:generation_tokens_total counter\n") + fmt.Fprintf(w, "vllm:generation_tokens_total{%s} %d\n", lbl, e.m.genTokens.Load()) + + fmt.Fprintf(w, "# HELP vllm:request_success_total Count of successfully processed requests.\n") + fmt.Fprintf(w, "# TYPE vllm:request_success_total counter\n") + fmt.Fprintf(w, "vllm:request_success_total{%s,finished_reason=\"length\"} %d\n", lbl, e.m.finished.Load()) + + // Simülasyona özgü (gerçek vLLM'de yok; kaos testlerinde işe yarar). + fmt.Fprintf(w, "# TYPE sim:requests_cancelled_total counter\n") + fmt.Fprintf(w, "sim:requests_cancelled_total{%s} %d\n", lbl, e.m.cancelled.Load()) + fmt.Fprintf(w, "# TYPE sim:requests_failed_total counter\n") + fmt.Fprintf(w, "sim:requests_failed_total{%s} %d\n", lbl, e.m.failed.Load()) + + e.m.mu.Lock() + defer e.m.mu.Unlock() + fmt.Fprintf(w, "# TYPE vllm:time_to_first_token_seconds histogram\n") + e.m.ttft.write(w, "vllm:time_to_first_token_seconds", lbl) + fmt.Fprintf(w, "# TYPE vllm:time_per_output_token_seconds histogram\n") + e.m.tpot.write(w, "vllm:time_per_output_token_seconds", lbl) +} diff --git a/internal/sim/server.go b/internal/sim/server.go new file mode 100644 index 0000000..b7039e0 --- /dev/null +++ b/internal/sim/server.go @@ -0,0 +1,358 @@ +package sim + +import ( + "encoding/json" + "fmt" + "log/slog" + "net/http" + "strings" + "time" +) + +// Server, Engine'i OpenAI uyumlu bir HTTP yüzeyiyle sarar. +type Server struct { + eng *Engine + log *slog.Logger + mux *http.ServeMux +} + +// NewServer, engine için HTTP yüzeyini kurar. +func NewServer(eng *Engine, log *slog.Logger) *Server { + if log == nil { + log = slog.Default() + } + s := &Server{eng: eng, log: log, mux: http.NewServeMux()} + s.mux.HandleFunc("POST /v1/chat/completions", s.handleChat) + s.mux.HandleFunc("POST /v1/completions", s.handleCompletions) + s.mux.HandleFunc("GET /v1/models", s.handleModels) + s.mux.HandleFunc("GET /health", s.handleHealth) + s.mux.HandleFunc("GET /metrics", s.handleMetrics) + return s +} + +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mux.ServeHTTP(w, r) } + +// --- istek/yanıt tipleri (yalnızca gereken alanlar) --- + +type streamOptions struct { + IncludeUsage bool `json:"include_usage"` +} + +type chatMessage struct { + Role string `json:"role"` + Content json.RawMessage `json:"content"` +} + +type chatRequest struct { + Model string `json:"model"` + Messages []chatMessage `json:"messages"` + MaxTokens int `json:"max_tokens"` + MaxCompletionTokens int `json:"max_completion_tokens"` + Stream bool `json:"stream"` + StreamOptions *streamOptions `json:"stream_options"` +} + +type completionRequest struct { + Model string `json:"model"` + Prompt json.RawMessage `json:"prompt"` + MaxTokens int `json:"max_tokens"` + Stream bool `json:"stream"` + StreamOptions *streamOptions `json:"stream_options"` +} + +type usage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` + // OpenAI uyumlu cache raporlaması; prefix cache etkisini istemciye gösterir. + PromptTokensDetails struct { + CachedTokens int `json:"cached_tokens"` + } `json:"prompt_tokens_details"` +} + +// text, mesaj içeriğini düz metne indirger. +// İçerik string ya da parça dizisi olabilir; sim için ikisini de kaba şekilde +// metne çeviriyoruz, çünkü tek ihtiyacımız deterministik bir token sayısı. +func (m chatMessage) text() string { + var s string + if err := json.Unmarshal(m.Content, &s); err == nil { + return s + } + return string(m.Content) +} + +func promptFromMessages(msgs []chatMessage) string { + var b strings.Builder + for _, m := range msgs { + b.WriteString(m.Role) + b.WriteString(": ") + b.WriteString(m.text()) + b.WriteString("\n") + } + return b.String() +} + +// --- handler'lar --- + +func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { + var req chatRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + maxTok := req.MaxTokens + if maxTok == 0 { + maxTok = req.MaxCompletionTokens + } + includeUsage := req.StreamOptions != nil && req.StreamOptions.IncludeUsage + + s.serve(w, r, servePlan{ + prompt: Prompt{Text: promptFromMessages(req.Messages), MaxTokens: maxTok}, + model: s.modelName(req.Model), + stream: req.Stream, + includeUsage: includeUsage, + object: "chat.completion", + chunkObject: "chat.completion.chunk", + chat: true, + }) +} + +func (s *Server) handleCompletions(w http.ResponseWriter, r *http.Request) { + var req completionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error()) + return + } + var prompt string + if err := json.Unmarshal(req.Prompt, &prompt); err != nil { + prompt = string(req.Prompt) + } + includeUsage := req.StreamOptions != nil && req.StreamOptions.IncludeUsage + + s.serve(w, r, servePlan{ + prompt: Prompt{Text: prompt, MaxTokens: req.MaxTokens}, + model: s.modelName(req.Model), + stream: req.Stream, + includeUsage: includeUsage, + object: "text_completion", + chunkObject: "text_completion", + chat: false, + }) +} + +func (s *Server) modelName(requested string) string { + if requested != "" { + return requested + } + return s.eng.Config().Model +} + +func (s *Server) handleModels(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{ + "object": "list", + "data": []map[string]any{{ + "id": s.eng.Config().Model, + "object": "model", + "created": time.Now().Unix(), + "owned_by": "llmgw-sim", + }}, + }) +} + +func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) +} + +func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + s.eng.WriteMetrics(w) +} + +// --- ortak akış --- + +type servePlan struct { + prompt Prompt + model string + stream bool + includeUsage bool + object string + chunkObject string + chat bool +} + +func (s *Server) serve(w http.ResponseWriter, r *http.Request, p servePlan) { + ch, err := s.eng.Submit(r.Context(), p.prompt) + if err != nil { + code := http.StatusInternalServerError + kind := "server_error" + if err == ErrTooLong { + code, kind = http.StatusBadRequest, "invalid_request_error" + } + writeError(w, code, kind, err.Error()) + return + } + + id := fmt.Sprintf("cmpl-sim-%d", time.Now().UnixNano()) + created := time.Now().Unix() + + if p.stream { + s.streamResponse(w, r, ch, p, id, created) + return + } + s.bufferedResponse(w, ch, p, id, created) +} + +func (s *Server) bufferedResponse(w http.ResponseWriter, ch <-chan Event, p servePlan, id string, created int64) { + var b strings.Builder + n := 0 + var res *Result + for ev := range ch { + switch ev.Kind { + case EventToken: + b.WriteString(ev.Text) + n++ + case EventError: + writeError(w, http.StatusInternalServerError, "server_error", ev.Err.Error()) + return + case EventDone: + res = ev.Result + } + } + + u := s.usage(p.prompt, n, res) + body := map[string]any{ + "id": id, "object": p.object, "created": created, "model": p.model, + "usage": u, + } + if p.chat { + body["choices"] = []map[string]any{{ + "index": 0, + "message": map[string]string{"role": "assistant", "content": b.String()}, + "finish_reason": "length", + }} + } else { + body["choices"] = []map[string]any{{ + "index": 0, "text": b.String(), "finish_reason": "length", + }} + } + writeJSON(w, http.StatusOK, body) +} + +func (s *Server) streamResponse(w http.ResponseWriter, r *http.Request, ch <-chan Event, p servePlan, id string, created int64) { + flusher, ok := w.(http.Flusher) + if !ok { + writeError(w, http.StatusInternalServerError, "server_error", "streaming desteklenmiyor") + return + } + h := w.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("Cache-Control", "no-cache") + h.Set("Connection", "keep-alive") + w.WriteHeader(http.StatusOK) + flusher.Flush() + + n := 0 + var res *Result + for { + select { + case <-r.Context().Done(): + // İstemci koptu. Engine reap() ile isteği batch'ten düşürecek. + return + case ev, open := <-ch: + if !open { + if p.includeUsage { + s.writeChunk(w, chunk(id, p, created, nil, "", s.usage(p.prompt, n, res))) + flusher.Flush() + } + fmt.Fprint(w, "data: [DONE]\n\n") + flusher.Flush() + return + } + switch ev.Kind { + case EventToken: + n++ + s.writeChunk(w, chunk(id, p, created, &ev.Text, "", nil)) + flusher.Flush() + case EventDone: + res = ev.Result + s.writeChunk(w, chunk(id, p, created, nil, "length", nil)) + flusher.Flush() + case EventError: + s.writeChunk(w, map[string]any{"error": map[string]any{ + "message": ev.Err.Error(), "type": "server_error", + }}) + flusher.Flush() + return + } + } + } +} + +func (s *Server) writeChunk(w http.ResponseWriter, v any) { + b, err := json.Marshal(v) + if err != nil { + s.log.Error("chunk serileştirilemedi", "err", err) + return + } + fmt.Fprintf(w, "data: %s\n\n", b) +} + +func chunk(id string, p servePlan, created int64, text *string, finish string, u any) map[string]any { + choice := map[string]any{"index": 0} + if p.chat { + delta := map[string]any{} + if text != nil { + delta["content"] = *text + } + choice["delta"] = delta + } else if text != nil { + choice["text"] = *text + } else { + choice["text"] = "" + } + if finish != "" { + choice["finish_reason"] = finish + } else { + choice["finish_reason"] = nil + } + + out := map[string]any{ + "id": id, "object": p.chunkObject, "created": created, "model": p.model, + } + // Usage chunk'ında choices boş dizi olur (OpenAI davranışı). + if u != nil { + out["choices"] = []map[string]any{} + out["usage"] = u + } else { + out["choices"] = []map[string]any{choice} + } + return out +} + +// usage, mümkünse engine'in raporladığı gerçek sayıları kullanır. +// res nil olabilir (istek iptal edildiyse veya hata aldıysa); o durumda +// prompt token sayısı metinden hesaplanır ve cached_tokens bilinmez. +func (s *Server) usage(p Prompt, completion int, res *Result) usage { + var u usage + u.PromptTokens = CountTokens(p.Text) + u.CompletionTokens = completion + if res != nil { + u.PromptTokens = res.PromptTokens + u.CompletionTokens = res.CompletionTokens + u.PromptTokensDetails.CachedTokens = res.CachedTokens + } + u.TotalTokens = u.PromptTokens + u.CompletionTokens + return u +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(v) +} + +func writeError(w http.ResponseWriter, code int, kind, msg string) { + writeJSON(w, code, map[string]any{"error": map[string]any{ + "message": msg, "type": kind, "code": nil, + }}) +} diff --git a/internal/sim/server_test.go b/internal/sim/server_test.go new file mode 100644 index 0000000..3cf064d --- /dev/null +++ b/internal/sim/server_test.go @@ -0,0 +1,383 @@ +package sim + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" +) + +func newTestServer(t *testing.T, cfg Config) *httptest.Server { + t.Helper() + eng := New(cfg) + srv := httptest.NewServer(NewServer(eng, nil)) + t.Cleanup(func() { + srv.Close() + eng.Close() + }) + return srv +} + +func postJSON(t *testing.T, url string, body any) *http.Response { + t.Helper() + b, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + resp, err := http.Post(url, "application/json", bytes.NewReader(b)) //nolint:noctx // test + if err != nil { + t.Fatal(err) + } + return resp +} + +func TestChatCompletionsBuffered(t *testing.T) { + srv := newTestServer(t, testConfig()) + + resp := postJSON(t, srv.URL+"/v1/chat/completions", map[string]any{ + "model": "sim-model", + "messages": []map[string]string{{"role": "user", "content": "merhaba"}}, + "max_tokens": 7, + }) + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("durum %d", resp.StatusCode) + } + + var out struct { + Object string `json:"object"` + Choices []struct { + Message struct { + Role string `json:"role"` + Content string `json:"content"` + } `json:"message"` + FinishReason string `json:"finish_reason"` + } `json:"choices"` + Usage usage `json:"usage"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatal(err) + } + + if out.Object != "chat.completion" { + t.Errorf("object=%q", out.Object) + } + if len(out.Choices) != 1 { + t.Fatalf("choices=%d", len(out.Choices)) + } + if n := len(strings.Fields(out.Choices[0].Message.Content)); n != 7 { + t.Errorf("beklenen 7 token, alınan %d", n) + } + if out.Usage.CompletionTokens != 7 { + t.Errorf("usage.completion_tokens=%d", out.Usage.CompletionTokens) + } + if out.Usage.TotalTokens != out.Usage.PromptTokens+out.Usage.CompletionTokens { + t.Error("usage toplamı tutarsız") + } +} + +// sseEvent, SSE akışından çözülmüş tek bir olay. +type sseEvent struct { + raw string + data map[string]any +} + +func readSSE(t *testing.T, body *bufio.Reader) []sseEvent { + t.Helper() + var out []sseEvent + for { + line, err := body.ReadString('\n') + if err != nil { + return out + } + line = strings.TrimRight(line, "\r\n") + if !strings.HasPrefix(line, "data: ") { + continue + } + payload := strings.TrimPrefix(line, "data: ") + ev := sseEvent{raw: payload} + if payload != "[DONE]" { + if err := json.Unmarshal([]byte(payload), &ev.data); err != nil { + t.Fatalf("SSE yükü çözülemedi: %v (%q)", err, payload) + } + } + out = append(out, ev) + } +} + +func TestChatCompletionsStreaming(t *testing.T) { + srv := newTestServer(t, testConfig()) + + resp := postJSON(t, srv.URL+"/v1/chat/completions", map[string]any{ + "model": "sim-model", + "messages": []map[string]string{{"role": "user", "content": "merhaba"}}, + "max_tokens": 5, + "stream": true, + "stream_options": map[string]any{"include_usage": true}, + }) + defer resp.Body.Close() + + if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" { + t.Errorf("Content-Type=%q", ct) + } + + events := readSSE(t, bufio.NewReader(resp.Body)) + if len(events) == 0 { + t.Fatal("hiç SSE olayı gelmedi") + } + if last := events[len(events)-1]; last.raw != "[DONE]" { + t.Errorf("akış [DONE] ile bitmedi: %q", last.raw) + } + + var contentTokens, usageChunks, finishChunks int + for _, ev := range events { + if ev.raw == "[DONE]" { + continue + } + if _, ok := ev.data["usage"]; ok { + usageChunks++ + continue + } + choices, _ := ev.data["choices"].([]any) + if len(choices) == 0 { + continue + } + c, _ := choices[0].(map[string]any) + if delta, ok := c["delta"].(map[string]any); ok { + if _, has := delta["content"]; has { + contentTokens++ + } + } + if fr, ok := c["finish_reason"].(string); ok && fr == "length" { + finishChunks++ + } + } + + if contentTokens != 5 { + t.Errorf("beklenen 5 içerik chunk'ı, alınan %d", contentTokens) + } + if finishChunks != 1 { + t.Errorf("beklenen 1 finish_reason chunk'ı, alınan %d", finishChunks) + } + if usageChunks != 1 { + t.Errorf("include_usage istendi ama %d usage chunk'ı geldi", usageChunks) + } +} + +func TestStreamingUsageIstenmezseGonderilmez(t *testing.T) { + srv := newTestServer(t, testConfig()) + + resp := postJSON(t, srv.URL+"/v1/chat/completions", map[string]any{ + "messages": []map[string]string{{"role": "user", "content": "x"}}, + "max_tokens": 2, + "stream": true, + }) + defer resp.Body.Close() + + for _, ev := range readSSE(t, bufio.NewReader(resp.Body)) { + if ev.raw == "[DONE]" { + continue + } + if _, ok := ev.data["usage"]; ok { + t.Error("istenmediği halde usage chunk'ı gönderildi") + } + } +} + +func TestIstemciKopmasiEngineeYansir(t *testing.T) { + // Faz 1'in çıkış kriterinin temeli: istemci koptuğunda GPU boşa çalışmamalı. + cfg := testConfig() + cfg.DecodeTokensPerSec = 20 + eng := New(cfg) + defer eng.Close() + srv := httptest.NewServer(NewServer(eng, nil)) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + body, _ := json.Marshal(map[string]any{ + "messages": []map[string]string{{"role": "user", "content": "merhaba"}}, + "max_tokens": 5000, + "stream": true, + }) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + srv.URL+"/v1/chat/completions", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + + // İlk token'ı bekle, sonra istemciyi kopar. + waitFor(t, 2*time.Second, func() bool { return eng.Stats().Running == 1 }) + cancel() + _ = resp.Body.Close() + + waitFor(t, 3*time.Second, func() bool { + s := eng.Stats() + return s.Running == 0 && s.KVCacheUsage == 0 + }) +} + +func TestCompletionsUcNoktasi(t *testing.T) { + srv := newTestServer(t, testConfig()) + + resp := postJSON(t, srv.URL+"/v1/completions", map[string]any{ + "model": "sim-model", "prompt": "merhaba", "max_tokens": 3, + }) + defer resp.Body.Close() + + var out struct { + Object string `json:"object"` + Choices []struct { + Text string `json:"text"` + } `json:"choices"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatal(err) + } + if out.Object != "text_completion" { + t.Errorf("object=%q", out.Object) + } + if n := len(strings.Fields(out.Choices[0].Text)); n != 3 { + t.Errorf("beklenen 3 token, alınan %d", n) + } +} + +func TestModelsVeHealth(t *testing.T) { + cfg := testConfig() + cfg.Model = "llama-3.3-70b" + srv := newTestServer(t, cfg) + + resp, err := http.Get(srv.URL + "/health") //nolint:noctx // test + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("/health durum %d", resp.StatusCode) + } + + resp, err = http.Get(srv.URL + "/v1/models") //nolint:noctx // test + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + var out struct { + Data []struct { + ID string `json:"id"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatal(err) + } + if len(out.Data) != 1 || out.Data[0].ID != "llama-3.3-70b" { + t.Errorf("/v1/models beklenen modeli listelemedi: %+v", out.Data) + } +} + +// TestMetriklerPrometheusIleParseEdilebilir, §7.6'daki engine kontratını doğrular. +// Router bu metrikleri gerçek bir Prometheus parser'ıyla okuyacak; format +// bozuksa yönlendirme kör kalır. +func TestMetriklerPrometheusIleParseEdilebilir(t *testing.T) { + srv := newTestServer(t, testConfig()) + + // Biraz trafik üret ki sayaçlar dolsun. + resp := postJSON(t, srv.URL+"/v1/chat/completions", map[string]any{ + "messages": []map[string]string{{"role": "user", "content": "merhaba dunya"}}, + "max_tokens": 4, + }) + resp.Body.Close() + + mresp, err := http.Get(srv.URL + "/metrics") //nolint:noctx // test + if err != nil { + t.Fatal(err) + } + defer mresp.Body.Close() + + // TextParser'ın sıfır değeri geçersiz; şema açıkça verilmeli. + // Legacy şema daha sıkı: metrik adlarımızın klasik Prometheus kurallarına + // uyduğunu da doğrulamış oluyoruz (vLLM adlarındaki ':' legacy'de de geçerli). + parser := expfmt.NewTextParser(model.LegacyValidation) + families, err := parser.TextToMetricFamilies(mresp.Body) + if err != nil { + t.Fatalf("metrikler parse edilemedi: %v", err) + } + + // §7.6 zorunlu üçlüsü. + for _, name := range []string{ + "vllm:num_requests_running", + "vllm:num_requests_waiting", + "vllm:kv_cache_usage_perc", + } { + if _, ok := families[name]; !ok { + t.Errorf("zorunlu metrik eksik: %s", name) + } + } + // Opsiyoneller ve histogramlar. + for _, name := range []string{ + "vllm:cache_config_info", + "vllm:prefix_cache_queries_total", + "vllm:prefix_cache_hits_total", + "vllm:time_to_first_token_seconds", + } { + if _, ok := families[name]; !ok { + t.Errorf("metrik eksik: %s", name) + } + } + + if f, ok := families["vllm:generation_tokens_total"]; ok { + if got := f.Metric[0].GetCounter().GetValue(); got != 4 { + t.Errorf("generation_tokens_total=%v, beklenen 4", got) + } + } else { + t.Error("vllm:generation_tokens_total eksik") + } +} + +// TestUsageCachedTokensRaporlanir, prefix cache etkisinin istemciye ulaştığını +// doğrular. Router ileride bunu kullanmayacak olsa da, cache'in gerçekten +// çalıştığını uçtan uca gösteren tek sinyal bu. +func TestUsageCachedTokensRaporlanir(t *testing.T) { + srv := newTestServer(t, testConfig()) + + body := map[string]any{ + "messages": []map[string]string{{"role": "user", "content": strings.Repeat("ortak onek. ", 100)}}, + "max_tokens": 3, + } + + // İlk istek cache'i doldurur. + resp := postJSON(t, srv.URL+"/v1/chat/completions", body) + resp.Body.Close() + + // İkincisi aynı ön eki kullanmalı. + resp = postJSON(t, srv.URL+"/v1/chat/completions", body) + defer resp.Body.Close() + + var out struct { + Usage usage `json:"usage"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatal(err) + } + if out.Usage.PromptTokensDetails.CachedTokens == 0 { + t.Error("ikinci istekte cached_tokens 0; prefix cache istemciye raporlanmıyor") + } + if out.Usage.PromptTokensDetails.CachedTokens > out.Usage.PromptTokens { + t.Errorf("cached_tokens (%d) prompt_tokens'tan (%d) büyük", + out.Usage.PromptTokensDetails.CachedTokens, out.Usage.PromptTokens) + } +} diff --git a/internal/sim/text.go b/internal/sim/text.go new file mode 100644 index 0000000..c302e66 --- /dev/null +++ b/internal/sim/text.go @@ -0,0 +1,26 @@ +package sim + +import "strings" + +// simWords, üretilen sahte çıktının kelime havuzu. +var simWords = []string{ + "alfa", "beta", "gama", "delta", "epsilon", "zeta", "eta", "teta", + "iota", "kappa", "lambda", "mu", "nu", "ksi", "omikron", "pi", +} + +// tokenText, n'inci üretilen token'ın metnini verir (n 1'den başlar). +// Deterministik olması şart: finish sırasında cache'e yazılan metinle +// istemciye akıtılan metin birebir aynı olmalı. +func tokenText(n int) string { + return " " + simWords[(n-1)%len(simWords)] +} + +// generatedText, ilk n token'ın birleşimini verir. +func generatedText(n int) string { + var b strings.Builder + b.Grow(n * 7) + for i := 1; i <= n; i++ { + b.WriteString(tokenText(i)) + } + return b.String() +} diff --git a/internal/sim/tokens.go b/internal/sim/tokens.go new file mode 100644 index 0000000..36a1de3 --- /dev/null +++ b/internal/sim/tokens.go @@ -0,0 +1,37 @@ +package sim + +import "hash/fnv" + +// TokensPerChar, sahte tokenizer'ın kaç bayta bir token ürettiği. +// Gerçek BPE tokenizer'lara kaba bir yaklaşım; simülasyon için tek gereken +// deterministik ve prefix-kararlı olması. +const bytesPerToken = 4 + +// Token, sahte tokenizer'ın ürettiği token kimliği. +type Token uint32 + +// Tokenize, metni deterministik ve prefix-kararlı şekilde token'lara böler. +// +// Prefix-kararlılık kritik: aynı ön eke sahip iki metnin token dizileri de aynı +// ön ekle başlamalı, yoksa prefix cache simülasyonu anlamsız olur. Bu yüzden +// bölme, metnin başından itibaren sabit bayt sayısıyla yapılır — içeriğe bağlı +// (BPE gibi) bir bölme kullanılmaz. +func Tokenize(s string) []Token { + if s == "" { + return nil + } + n := (len(s) + bytesPerToken - 1) / bytesPerToken + out := make([]Token, 0, n) + for i := 0; i < len(s); i += bytesPerToken { + end := min(i+bytesPerToken, len(s)) + h := fnv.New32a() + _, _ = h.Write([]byte(s[i:end])) + out = append(out, Token(h.Sum32())) + } + return out +} + +// CountTokens, metnin token sayısını tokenize etmeden verir. +func CountTokens(s string) int { + return (len(s) + bytesPerToken - 1) / bytesPerToken +} diff --git a/internal/sim/tokens_test.go b/internal/sim/tokens_test.go new file mode 100644 index 0000000..cd544fa --- /dev/null +++ b/internal/sim/tokens_test.go @@ -0,0 +1,24 @@ +package sim + +import "testing" + +func TestTokenizeOnEkKararli(t *testing.T) { + a := Tokenize("merhaba") + b := Tokenize("merhaba dunya") + if len(a) > len(b) { + t.Fatal("uzun metin daha az token verdi") + } + for i := range a[:len(a)-1] { // son token yarım olabilir + if a[i] != b[i] { + t.Fatalf("token %d ön ek kararlı değil", i) + } + } +} + +func TestCountTokensTokenizeIleTutarli(t *testing.T) { + for _, s := range []string{"", "a", "abcd", "abcde", "uzunca bir prompt metni"} { + if got, want := CountTokens(s), len(Tokenize(s)); got != want { + t.Errorf("%q: CountTokens=%d Tokenize=%d", s, got, want) + } + } +} diff --git a/internal/version/version.go b/internal/version/version.go new file mode 100644 index 0000000..dffa08c --- /dev/null +++ b/internal/version/version.go @@ -0,0 +1,5 @@ +// Package version, build sırasında ldflags ile doldurulan sürüm bilgisini tutar. +package version + +// Version, `make build` tarafından git describe çıktısıyla doldurulur. +var Version = "dev" diff --git a/llmgw-plan.md b/llmgw-plan.md new file mode 100644 index 0000000..ec6b580 --- /dev/null +++ b/llmgw-plan.md @@ -0,0 +1,1045 @@ +# llmgw — Çoklu IB Fabric ve Heterojen GPU için Tek Binary LLM Gateway + +> Durum: Tasarım / Plan v0.2 +> v0.2 değişikliği: mevcut LLM gateway/router projeleri incelendi (§2.5) ve kritik +> tasarımları plana alındı — en önemlisi **admission ile scheduling'in ayrılması** (§7.2), +> eklenti tabanlı **filter → score → pick** çerçevesi (§7.3) ve **prefix/KV cache +> farkındalığının üç kademeli** modeli (§7.5). +> Hedef: Tek bir Go binary'si ile (farklı roller altında) kendi GPU sunucularımızda çalışan +> LLM'leri tek bir OpenAI uyumlu uç noktadan sunmak. Arkada birden fazla bağımsız +> InfiniBand fabric, IB'siz node'lar ve farklı GPU tipleri var. + +--- + +## 1. Kapsam + +### 1.1 Yapılacaklar +- OpenAI uyumlu API (`/v1/chat/completions`, `/v1/completions`, `/v1/embeddings`, `/v1/models`), SSE streaming dahil. +- Her GPU node'unda çalışan bir **agent**: GPU ve IB keşfi, inference engine süreçlerini (vLLM / SGLang) başlatma, izleme, durdurma. +- **Control plane**: node envanteri, model kataloğu, replika yerleşimi (placement), durum takibi. +- **Router**: iki katmanlı — **admission** (öncelik bantları, tenant adaleti, doygunluk kuyruklaması) + ve **scheduling** (eklenti tabanlı filter/score/pick; kuyruk/KV-cache/prefix farkındalığı), + üstüne fallback, retry, cooldown, circuit breaker. +- Fabric farkındalığı: multi-node replika **asla** iki fabric'e yayılmaz; TP grubu **asla** farklı GPU tiplerini karıştırmaz. +- API key, rate limit, kullanım (token) muhasebesi, Prometheus metrikleri. + +### 1.2 Yapılmayacaklar (en azından v1'de) +- Inference'ın kendisi. Go sadece orkestrasyon ve proxy yapar; modeli vLLM/SGLang çalıştırır. +- Kubernetes operatörü (bare metal / systemd hedefleniyor; K8s sonra düşünülebilir). +- Prefill/decode ayrıştırması (v2 hedefi, bkz. §12). +- Dış provider'lar (OpenAI, Anthropic) — mimari buna açık bırakılacak ama v1'de yok. + +--- + +## 2. Terimler + +| Terim | Anlamı | +|---|---| +| **Node** | Agent'ın çalıştığı fiziksel sunucu. | +| **Fabric / Ada (Island)** | Bağımsız bir IB fabric'i. IB'siz node'lar `none` adasındadır ve her biri kendi başına tek-node adası sayılır. | +| **GPU Class** | Aynı model + aynı VRAM'e sahip GPU grubu (ör. `h100-80g`, `a100-40g`, `l40s-48g`). | +| **Pool** | (fabric, gpu class) ikilisi. Yerleşim birimi. | +| **Model** | Katalogdaki mantıksal model (ör. `llama-3.3-70b`). Kullanıcı bu adı görür. | +| **Profile** | Bir modelin belirli bir GPU class üzerinde nasıl çalışacağı (TP, PP, quant, max-len, engine argümanları). | +| **Replica** | Bir modelin çalışan tek bir kopyası. 1..N node'a yayılabilir. Router'ın gördüğü birim budur. | +| **Shard** | Bir replikanın tek bir node'daki parçası (engine süreci). | +| **Yönlendirme Profili** | Router'ın bir istek sınıfı için kullandığı scorer seti + ağırlıkları (chat / batch / long-context). §2'deki model **Profile**'ı ile karıştırma. | +| **Band** | Öncelik sınıfı (platinum/gold/silver/bronze). Admission kararının birimi. | +| **Block** | KV cache'in muhasebe birimi; sabit token sayısı (vLLM'de 16, GIE indexer'ında varsayılan 64). | + +--- + +## 2.5 Referans Uygulamalar ve Ne Aldık + +Sıfırdan tasarlamak yerine bu alanda olgunlaşmış projelerin çözdüğü problemleri devraldık. + +| Proje | Konumu | llmgw'ye aldığımız | +|---|---|---| +| **Gateway API Inference Extension (EPP)** | K8s standardı; Envoy `ext_proc` üzerinden çalışan Endpoint Picker | Admission/scheduling ayrımı, eklenti tabanlı `filter → score → pick` çerçevesi, **model server metrik kontratı** (§7.6) | +| **llm-d** | GIE üzerinde "well-lit path"ler | KV event tabanlı **kesin** prefix index, öncelik bantları + tenant adaleti, doygunluk dedektörü | +| **AIBrix** (vLLM projesi) | Envoy ext_proc gateway + heterojen GPU optimizer | **Heterojen GPU SLO/maliyet modeli** (§6.2.1), gateway'de kuyruklama (`slo-least-load-pulling`), yük dengesizliğinde prefix affinity'yi iptal etme | +| **NVIDIA Dynamo** | Radix-tree KV router + SLA planner | Maliyet fonksiyonunun formu (§7.4), `router_temperature` ile sürü etkisine karşı softmax örnekleme | +| **SGLang router** | Yaklaşık radix tree ile cache-aware LB | v1'in prefix çözümü: yaklaşık ağaç + eşik mantığı (§7.5 Kademe 1) | +| **vLLM production-stack / vllm-project/router** | vLLM'in kendi router'ları | Session affinity (ölçümlerde %96 vs %93 cache hit), consistent hashing | +| **LiteLLM** | En yaygın çok-provider proxy'si | Virtual key → takım/proje → bütçe hiyerarşisi, cooldown, spend yazımının istek yolundan çıkarılması, bütçe aşımında fallback | +| **Envoy AI Gateway** | Envoy tabanlı AI gateway (v1.0, Haziran 2026) | Token bazlı (sadece istek bazlı değil) rate limit | + +Bilerek **almadıklarımız:** Kubernetes bağımlılığı (bare metal hedefi), Envoy `ext_proc` mimarisi +(tek binary hedefi — router'ı doğrudan Go'da yazıyoruz), semantic router / LLM tabanlı model seçimi +(kapsam dışı), dış provider yönetimi (v1'de yok). + +--- + +## 3. Mimari + +``` + ┌───────────────────────────────────────────┐ + İstemciler ─────► │ llmgw server │ + (OpenAI API) │ ┌────────────┐ ┌──────────────────────┐ │ + │ │ Router │ │ Control Plane │ │ + │ │ (data) │◄─┤ - Inventory │ │ + │ │ │ │ - Catalog │ │ + │ └─────┬──────┘ │ - Scheduler │ │ + │ │ │ - Store (SQLite) │ │ + │ │ └─────────▲────────────┘ │ + └────────┼───────────────────┼──────────────┘ + HTTP (eth) │ │ gRPC stream (mTLS, agent başlatır) + ┌────────────────────┼───────────────────┼─────────────────────┐ + ▼ ▼ │ ▼ + ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ + │ llmgw │ IB-A │ llmgw │ │ llmgw │ IB-B │ llmgw │ + │ agent │◄──────►│ agent │ │ agent │◄───────►│ agent │ ... + │ vLLM rank0│ NCCL │ vLLM rank1│ │ SGLang │ │ (IB yok) │ + └───────────┘ └───────────┘ └───────────┘ └───────────┘ + Fabric A adası Fabric B adası none adası +``` + +### 3.1 Tek binary, çok rol + +``` +llmgw server --config server.yaml # router + control plane (v1: aynı süreç) +llmgw agent --config agent.yaml # her GPU node'unda +llmgw ctl ... # CLI: node listele, model deploy, drain vs. +llmgw sim-engine # test için sahte OpenAI uyumlu engine +``` + +`router` ve `control` ileride ayrı süreçler olarak da çalışabilmeli (`llmgw server --roles=router`), bu yüzden aralarındaki bağımlılık bir interface üzerinden olmalı (bkz. §6.4). + +### 3.2 Build notu +`go-nvml` cgo gerektirir. Agent'taki NVML kodunu `//go:build nvml` tag'i arkasına koy; tag'siz build'de sahte/`nvidia-smi` tabanlı keşif kullanılsın. Böylece server tarafı cgo'suz derlenebilir ve geliştirme makinesinde GPU olmadan test yapılabilir. + +--- + +## 4. Proje Yapısı + +``` +llmgw/ +├── cmd/llmgw/main.go # cobra root; alt komutlar +├── api/ +│ └── proto/llmgw/v1/*.proto # agent <-> control plane +├── internal/ +│ ├── config/ # koanf ile YAML + env +│ ├── agent/ +│ │ ├── agent.go # ana döngü, control plane bağlantısı +│ │ ├── discovery/ +│ │ │ ├── gpu_nvml.go # //go:build nvml +│ │ │ ├── gpu_fallback.go +│ │ │ ├── ib.go # /sys/class/infiniband okuma +│ │ │ └── topology.go # NVLink, PCIe, NUMA, GPU<->HCA yakınlığı +│ │ ├── supervisor/ # engine süreç yönetimi (exec, restart, log) +│ │ ├── engines/ +│ │ │ ├── engine.go # Engine interface +│ │ │ ├── vllm.go +│ │ │ └── sglang.go +│ │ └── health/ # engine /health ve /metrics yoklama +│ ├── control/ +│ │ ├── inventory/ # node & GPU durumu +│ │ ├── catalog/ # model + profile tanımları +│ │ ├── scheduler/ # placement algoritması +│ │ ├── reconciler/ # desired vs actual döngüsü +│ │ └── agentsvc/ # gRPC server +│ ├── router/ +│ │ ├── openai/ # HTTP handler'lar, request parse +│ │ ├── proxy/ # upstream'e stream proxy +│ │ ├── admission/ # doygunluk dedektörü, bant kuyrukları (§7.2) +│ │ ├── scheduler/ # Filter/Scorer/Picker çerçevesi (§7.3) +│ │ │ ├── filters/ +│ │ │ ├── scorers/ # queue, kv, prefix, load, inflight, session +│ │ │ └── pickers/ # max-score, p2c, softmax +│ │ ├── prefix/ # Kademe 0 session hash, Kademe 1 approx tree, Kademe 2 kv index +│ │ ├── breaker/ # circuit breaker + cooldown +│ │ └── metricsscrape/ # engine /metrics okuma (§7.6 kontratı) +│ ├── auth/ # API key, org/takım/proje, bütçe +│ ├── ratelimit/ # token bucket (istek + token) +│ ├── usage/ # token sayımı, kayıt +│ ├── store/ # SQLite (modernc.org/sqlite), migration +│ ├── obs/ # slog, prometheus, otel +│ └── sim/ # sahte engine +├── deploy/ +│ ├── systemd/ +│ └── examples/ +├── test/e2e/ +└── docs/ +``` + +### 4.1 Kütüphaneler + +| İhtiyaç | Seçim | +|---|---| +| CLI | `github.com/spf13/cobra` | +| Config | `github.com/knadh/koanf/v2` | +| RPC | `connectrpc.com/connect` (gRPC uyumlu, HTTP/2, debug kolay) | +| GPU | `github.com/NVIDIA/go-nvml` | +| DB | `modernc.org/sqlite` (cgo'suz) + `pressly/goose` migration | +| Metrik | `github.com/prometheus/client_golang`, parse için `github.com/prometheus/common/expfmt` | +| Tracing | `go.opentelemetry.io/otel` | +| Log | `log/slog` (stdlib) | +| Rate limit | `golang.org/x/time/rate` (+ token bazlı için kendi bucket'ımız) | +| Hash | `github.com/cespare/xxhash/v2` | +| HA (v2) | `github.com/hashicorp/raft` veya etcd | +| Test | stdlib `testing`, `github.com/stretchr/testify` (opsiyonel) | + +--- + +## 5. Agent + +### 5.1 GPU keşfi +Her GPU için: +- index, UUID, isim, VRAM, compute capability (FP8 vb. desteği için), sürücü/CUDA sürümü +- NUMA node, PCIe bus id +- NVLink eşleri (NVML `GetNvLinkRemotePciInfo`) → aynı node içinde TP için hangi GPU'lar birlikte verimli +- sağlık: XID hataları, ECC, sıcaklık, kullanımdaki bellek (engine dışı süreçler var mı?) + +`GPUClass` otomatik türetilir: `normalize(name) + "-" + round(vram_gb) + "g"` → `h100-80g`. Config ile override edilebilir. + +### 5.2 IB keşfi ve fabric kimliği +`/sys/class/infiniband//ports//` altından: +- `state` (ACTIVE?), `phys_state`, `rate`, `link_layer` (InfiniBand / Ethernet → RoCE) +- `gids/0` → subnet prefix +- `sm_lid`, `lid` +- HCA'nın PCIe adresi ve NUMA'sı → GPU yakınlığı + +**Fabric kimliği nasıl belirlenir?** Otomatik tespit güvenilir değildir çünkü OpenSM varsayılan subnet prefix'i (`fe80::`) tüm fabric'lerde aynıdır. Öncelik sırası: + +1. Agent config'inde açık etiket: `fabrics: [{name: ib-a, hcas: [mlx5_0, mlx5_1]}]` **(önerilen)** +2. Her OpenSM'de benzersiz `subnet_prefix` ayarlanmışsa GID prefix'inden otomatik eşleme (server config'inde `prefix → fabric` tablosu) +3. Hiçbiri yoksa: node `fabric=unknown` olarak işaretlenir ve **multi-node yerleşime alınmaz**. + +Bonus doğrulama (v1.1): aynı fabric'te olduğunu iddia eden iki node arasında agent'lar `ibv_rc_pingpong` benzeri kısa bir test veya `ibping` ile erişilebilirliği doğrular. Tutarsızlık alarm üretir. + +Bir node'da birden fazla fabric'e bağlı HCA olabilir; model buna izin vermeli (`Node.Fabrics []FabricAttachment`). Ancak bir replika tek bir fabric seçer ve sadece o fabric'in HCA'larını kullanır. + +### 5.3 Engine interface + +```go +type Engine interface { + Name() string // "vllm", "sglang" + Validate(p Profile, gpus []GPU) error // compute capability, quant uyumu vs. + BuildCommand(spec ShardSpec) (*exec.Cmd, error) // argüman + env üretimi + HealthURL(spec ShardSpec) string + MetricsURL(spec ShardSpec) string + ParseMetrics(r io.Reader) (EngineStats, error) // engine'e özgü metrik adları +} + +type ShardSpec struct { + ReplicaID string + Rank int // 0 = leader (HTTP'yi sunan) + NNodes int + LeaderAddr string // dist init adresi (ethernet IP) + LeaderPort int + GPUIndices []int + HCAs []string // seçilen fabric'in HCA'ları + HTTPPort int + Profile Profile + ModelPath string +} + +type EngineStats struct { + // §7.6'daki zorunlu kontrat — bunlar olmadan router yönlendiremez + Running, Waiting int + KVCacheUsage float64 // 0..1 (vllm:kv_cache_usage_perc) + // opsiyonel + BlockSize, NumGPUBlocks int + PrefixHitRate float64 + TTFTp50, TPOTp50 time.Duration // histogramdan + DrainRate float64 // gözlenen tamamlanma/sn; `load` scorer'ı için + LoRA LoRAState // v2 + ScrapedAt time.Time // stale tespiti için — router bunu kullanır +} +``` + +Yeni bir engine eklemenin tanımı = §7.6'daki zorunlu üçlüyü doldurabilmek. + +**Multi-node başlatma:** +- SGLang: `--nnodes N --node-rank R --dist-init-addr host:port --tp T` doğal olarak destekler. +- vLLM: kullanılan sürüme göre ya Ray (leader'da `ray start --head`, worker'larda `ray start --address`) ya da yeni sürümlerdeki çok-node multiprocessing bayrakları. **Kullanacağın vLLM sürümünün dokümanını kontrol et**; `vllm.go` içinde iki yol da `launchMode` ile seçilebilir olsun. + +**Env üretimi (kritik):** +``` +CUDA_VISIBLE_DEVICES= +NCCL_IB_HCA= # sadece seçilen fabric +NCCL_SOCKET_IFNAME= # bootstrap +NCCL_IB_DISABLE=1 # tek-node veya IB'siz ise +NCCL_CROSS_NIC / NCCL_NET_GDR_LEVEL # profile'dan opsiyonel +GLOO_SOCKET_IFNAME= +``` + +### 5.4 Supervisor +- Süreçleri kendi process group'unda başlat (`SysProcAttr.Setpgid`), durdururken tüm grubu öldür (SIGTERM → timeout → SIGKILL). +- stdout/stderr → satır bazlı slog'a + halka tampon (son N satır, `llmgw ctl logs` için). +- Durum makinesi: `Pending → Starting → Loading → Ready → Draining → Stopped | Failed`. +- `Ready` ancak `/health` 200 ve `/v1/models` modeli listeliyorsa. +- Model yükleme uzun sürer; `startupTimeout` profile'dan gelir (ör. 20 dk). +- Çökme: control plane'e raporla, yeniden başlatma kararını **control plane** verir (multi-node replikada tek shard'ı yeniden başlatmak işe yaramaz; tüm replika yeniden kurulmalı). +- Agent yeniden başlarsa: çalışan engine süreçlerini (pid dosyası + cmdline kontrolü) sahiplenmeyi dene, yoksa temizle. + +### 5.5 Model ağırlıkları +- v1: paylaşımlı dosya sistemi (NFS/Lustre/Weka) veya node'da önceden indirilmiş yol. Agent sadece varlığını ve boyutunu kontrol eder. +- v1.1: agent'ın HF/S3'ten yerel cache'e indirmesi, ilerleme raporlaması, disk kotası. + +### 5.6 Agent ↔ Control plane protokolü +Agent bağlantıyı başlatır (NAT/firewall dostu), çift yönlü stream: + +```proto +service AgentService { + rpc Connect(stream AgentMessage) returns (stream ControlMessage); +} + +message AgentMessage { + oneof msg { + Register register = 1; // node bilgisi, GPU'lar, fabric'ler, sürümler + Heartbeat heartbeat = 2; // her 5 sn: GPU sağlığı, shard durumları + ShardStatus shard = 3; // durum değişimi anında + EngineStats stats = 4; // her 2 sn (router da doğrudan scrape edebilir) + } +} + +message ControlMessage { + oneof msg { + StartShard start = 1; + StopShard stop = 2; + DrainShard drain = 3; + Ack ack = 4; + } +} +``` + +- Komutlar idempotent: `StartShard` aynı `shard_id` ile tekrar gelirse no-op. +- mTLS: node kimliği sertifika CN'sinden. v1'de basit bir iç CA; `llmgw ctl cert issue`. +- Heartbeat 3 kez kaçarsa node `Unreachable`, 60 sn sonra `Lost` → üzerindeki replikalar yeniden planlanır. + +--- + +## 6. Control Plane + +### 6.1 Veri modeli (özet) + +```go +type Node struct { + ID, Hostname, MgmtAddr string + Labels map[string]string + GPUs []GPU + Fabrics []FabricAttachment // boşsa IB'siz + State NodeState // Ready, Cordoned, Draining, Unreachable, Lost + LastSeen time.Time +} + +type FabricAttachment struct { + Fabric string // "ib-a" + HCAs []string + Rate int // Gbps +} + +type Model struct { + Name string // kullanıcıya görünen + Aliases []string + Source string // yol veya HF id + Profiles []Profile + Scaling Scaling // min/max replika, hedef kuyruk vs. + Fallbacks []string // başka model adları +} + +type Profile struct { + Name string + GPUClass string // "h100-80g" + Engine string // "vllm" + TP, PP int + GPUsPerNode int // genelde TP + Nodes int // PP*... ; >1 ise fabric zorunlu + Args []string // --quantization fp8 --max-model-len 32768 ... + Env map[string]string + Weight int // router için kapasite ağırlığı (benchmark'tan) + Priority int // yerleşim tercihi + StartupTimeout time.Duration +} + +type Replica struct { + ID, Model, Profile string + Fabric string // multi-node ise dolu + Shards []Shard + State ReplicaState + Endpoint string // rank0 HTTP adresi +} +``` + +### 6.2 Scheduler (placement) + +Girdi: istenen replika sayısı, profile listesi, envanter. +Kısıtlar (**hard**): +1. Tüm shard'lar aynı `GPUClass`. +2. `Nodes > 1` ise tüm node'lar aynı `Fabric`'e bağlı olmalı ve o fabric'te ACTIVE HCA'ları olmalı. +3. `GPUsPerNode` kadar boş GPU; tercihen aynı NVLink domain'inde ve HCA ile aynı NUMA'da. +4. Node `Ready` ve `Cordoned` değil. +5. Engine `Validate` geçmeli (ör. FP8 profile'ı compute capability < 8.9 olan karta konmaz). + +Tercihler (**soft**, skor): +- Profile `Priority` +- Bin-packing: kısmen dolu node'ları önce doldur (büyük modeller için boş node'ları koru) +- Aynı modelin replikalarını farklı fabric/node'lara yay (hata alanı dağıtımı) +- GPU-HCA NUMA yakınlığı + +Algoritma (v1, basit ve deterministik): + +``` +for each profile in model.Profiles sorted by Priority desc: + pools := groupNodes(by GPUClass == profile.GPUClass, by Fabric if profile.Nodes > 1) + for each pool sorted by score: + nodes := pickNodes(pool, profile.Nodes, profile.GPUsPerNode) // greedy + NVLink grubu seçimi + if nodes != nil: + return Placement{profile, pool.Fabric, nodes, gpuAssignments} +return ErrUnschedulable (sebepleriyle birlikte — debug için çok önemli) +``` + +`groupNodes`: IB'siz node'lar `Nodes == 1` profile'ları için havuza girer; `Nodes > 1` için asla. + +`ctl` komutu: `llmgw ctl explain-placement ` her havuzun neden elendiğini göstermeli. + +### 6.2.1 Heterojen GPU: SLO/kapasite modeli + +Farklı GPU sınıflarını "ağırlık" gibi tek bir skalerle karşılaştırmak yetersiz — AIBrix'in +heterojen GPU optimizer'ından alınan model daha doğru. Her (model, profile) çifti için +`ctl bench` (§10) bir **kapasite profili** üretir: + +```go +type CapacityProfile struct { + Model, Profile string + GPUClass string + // SLO'yu karşılayan maksimum eşzamanlılık; asıl kapasite ölçüsü + MaxConcurrencyAtSLO int + ThroughputTokPerSec float64 + TTFTp99, TPOTp99 time.Duration + RelativeCost float64 // GPU-saat cinsinden göreli maliyet (config'ten) + MeasuredAt time.Time + EngineVersion string // sürüm değişince profil geçersiz +} +``` + +SLO örneği: `TPOT p99 ≤ 50ms && TTFT p99 ≤ 2s`. + +Bu profil üç yerde kullanılır: +1. **Router ağırlığı:** `Profile.Weight` artık elle yazılmaz, `MaxConcurrencyAtSLO`'dan türetilir. +2. **Admission eşikleri:** `TargetWaitingPerReplica` (§7.2.1) ölçülen değerden önerilir. +3. **Yerleşim/autoscaling hedefi:** "SLO'yu karşılayarak toplam maliyeti minimize et". Tahmini + yük için gereken replika sayısı GPU sınıfı başına `ceil(load / MaxConcurrencyAtSLO)` ile + hesaplanır; maliyeti en düşük kombinasyon seçilir. v1'de bu sadece `ctl` ile öneri üretir + (`ctl plan --target-rps N`), otomatik uygulama v1.1 autoscaler'ında. + +`EngineVersion` veya profile argümanları değiştiğinde ölçüm bayatlar: llmgw bunu uyarı olarak +yüzeye çıkarmalı (`llmgw_capacity_profile_stale{model,profile}`), sessizce eski sayıyı kullanmamalı. + +### 6.3 Reconciler +- Her 5 sn veya olay geldiğinde: desired (model scaling) ile actual (replica durumları) karşılaştır. +- Eksik replika → schedule → shard'lar için `StartShard` gönder (önce worker'lar, sonra rank0 veya engine'in istediği sıra). +- Replika içindeki herhangi bir shard `Failed` → tüm replikayı durdur, backoff ile yeniden planla. +- Fazla replika → önce `Drain` (router yeni istek göndermez, mevcutlar biter), sonra `Stop`. +- Node `Lost` → replikalarını `Failed` say. + +Autoscaling (v1.1): model başına `targetWaitingPerReplica`; router metriklerinden ortalama hesaplanır, cooldown'lı ölçekleme. Boşta GPU yoksa düşük öncelikli modelleri küçültme (preemption) v2. + +### 6.4 Router ile arayüz + +```go +type EndpointSource interface { + // model adı -> Ready replikalar (+ pool, weight, fabric, gpuclass, capacity profile) + Endpoints(model string) []Endpoint + Subscribe() <-chan struct{} // değişiklik bildirimi +} +``` +v1'de bellek içi; router ve control ayrılırsa aynı interface bir watch stream'i ile gerçeklenir. + +Router bu listeyi **her zaman önbellekte tutar**: control plane düşse bile mevcut replikalara +yönlendirmeye devam eder (§14'teki SPOF riskinin azaltılması). Önbellek yaşlandığında +(`endpoint_cache_max_age`, ör. 5 dk) sadece uyarı metriği üretir, trafiği kesmez. + +### 6.5 Store +- SQLite, WAL modu. Tablolar: `nodes`, `models`, `profiles`, `replicas`, `shards`, `api_keys`, `usage_rollup`, `events`. +- Envanter ve replika durumu aslında agent'lardan yeniden kurulabilir; kalıcı olması gereken asıl şey katalog, anahtarlar ve kullanım. +- HA (v2): Raft ile replike edilmiş durum veya dış Postgres/etcd. + +--- + +## 7. Router + +### 7.1 İstek yaşam döngüsü + +``` +İstek + ├─ 1. Auth : key → proje/takım, izinli modeller, öncelik bandı, yönlendirme profili + ├─ 2. Kota : RPM/TPM + bütçe (bellek içi sayaç; DB istek yolunda değil) + ├─ 3. Parse : model, stream, messages/prompt, max_tokens, user. + │ Gövdenin kalanı JSON olarak dokunulmadan iletilir. + ├─ 4. Çözümleme : model/alias → aday havuzlar (fabric × gpu_class) + ├─ 5. ADMISSION : "şimdi mi?" §7.2 ← havuz doygunsa istek burada bekler + ├─ 6. SCHEDULING : "nereye?" §7.3 filter → score → pick + ├─ 7. Proxy : upstream'e stream, iptal propagasyonu §7.8 + └─ 8. İstek sonrası (asenkron): usage, spend, metrik, log +``` + +**Kritik ayrım (GIE / llm-d'den alındı).** *Ne zaman* ilerleyeceği ile *nereye* gideceği iki ayrı +karardır ve ayrı tutulmalıdır. Ayrı tutulduğunda router istekleri kendi merkezî kuyruğunda tutup +öncelik ve tenant adaletini **istek engine kuyruğuna girmeden önce** uygulayabilir. + +Neden önemli: engine'in kendi kuyruğu FIFO'dur ve tenant/öncelik kavramı yoktur. İstek oraya bir +kez girdikten sonra hiçbir önceliklendirme yapılamaz. Bir batch işi 500 istekle vLLM kuyruğunu +doldurduğunda, arkadan gelen interaktif sohbet isteğinin TTFT'si çöker ve router'ın yapabileceği +hiçbir şey kalmaz. v0.1'de bu katman yoktu; **paylaşımlı bir küme için en kritik eksikti.** + +### 7.2 Admission control (Flow Control) + +#### 7.2.1 Doygunluk dedektörü + +Havuz başına tek bir doygunluk sinyali üretilir; iki metriği birleştirir (llm-d'nin yaklaşımı — +tek başına kuyruk derinliği yetmez, KV baskısı da gerçek bir tükenme sinyalidir): + +```go +// replika başına, sonra havuz için p90 veya ortalama +func saturation(e EngineStats, t Thresholds) float64 { + q := float64(e.Waiting) / t.TargetWaitingPerReplica // ör. hedef 4 + kv := e.KVCacheUsage / t.KVCeiling // ör. tavan 0.90 + return math.Max(q, kv) +} +``` + +Hedef, engine'de **küçük ama sağlıklı bir tampon** tutmaktır: continuous batching'in verimli +çalışmasına yetecek kadar istek olsun, ama kuyruk şişmesin. `TargetWaitingPerReplica` çok +düşük olursa GPU aç kalır, çok yüksek olursa kuyruk gecikmesi geri gelir. Bu iki eşik +model/profile başına ayarlanabilir olmalı ve `ctl bench` (§10) bunları önermeli. + +#### 7.2.2 Öncelik bantları + +Her API key bir banda eşlenir. Bant, doygunluk tavanını belirler: + +```yaml +bands: + - { name: platinum, priority: 400, saturation_ceiling: 1.00 } # hiç bekletme + - { name: gold, priority: 300, saturation_ceiling: 0.85 } + - { name: silver, priority: 200, saturation_ceiling: 0.60 } + - { name: bronze, priority: 100, saturation_ceiling: 0.30 } # batch işleri +``` + +Dispatch döngüsü bantları yüksekten düşüğe tarar. Havuzun doygunluğu bandın tavanını aştıysa +o bandın istekleri kuyrukta bekler. Sonuç: küme dolarken önce batch trafiği yavaşlar, interaktif +trafik korunur. (llm-d ölçümlerinde platinum/gold/silver 1 sn altında kalırken bronze batch +belirgin şekilde geriye düşüyor — istenen davranış tam olarak bu.) + +#### 7.2.3 Adalet ve kuyruk yönetimi + +- **Bant içi adalet:** aynı bantta birden fazla tenant varsa dispatch turları round-robin dağıtılır. + Tek bir tenant'ın ani yük artışı bandın tüm kapasitesini yutamaz. +- **Kapasite:** bant başına kuyruk boyu sınırlı. Dolduğunda yeni istek anında `503` (kuyrukta + sessizce birikmesindense hızlı reddedilsin). +- **TTL:** kuyrukta `queue_ttl` süresini aşan istek düşürülür (`504`). İstemcinin zaten vazgeçtiği + bir isteği GPU'ya göndermenin anlamı yok. +- **İstemci kopması:** kuyruktayken bağlantı koparsa istek sessizce düşer (context iptali). +- **Displacement (v1.1):** yüksek banda yer açmak için en düşük banttaki **en yeni** istekler + düşürülür (en eskiler zaten TTL'e yakın ve beklemeye en çok yatırım yapmış olanlar). + +```go +type Admission interface { + // İstek ilerleyebilene kadar bloke olur; ctx iptal/TTL'de hata döner. + Acquire(ctx context.Context, req *RequestInfo) (Lease, error) +} +``` + +`Lease.Release()` yanıt tamamlandığında çağrılır; router'ın in-flight sayacını da bu tutar. + +### 7.3 Scheduler: filter → score → pick + +v0.1'deki "strateji" listesi yerine eklenti çerçevesi (GIE EPP modeli). Her strateji ayrı bir +monolit olmaz; scorer'lar ağırlıklandırılarak birleştirilir. + +```go +type Filter interface { // adayı eler (hard constraint) + Name() string + Filter(ctx *SchedCtx, eps []Endpoint) []Endpoint +} + +type Scorer interface { // 0..1 arası skor; 1 = en iyi + Name() string + Score(ctx *SchedCtx, eps []Endpoint) map[EndpointID]float64 +} + +type Picker interface { // skorlardan seçim + Pick(ctx *SchedCtx, scores map[EndpointID]float64) Endpoint +} + +type RoutingProfile struct { + Name string + Filters []Filter + Scorers []WeightedScorer // {Scorer, Weight} + Picker Picker +} +``` + +Toplam skor: `score(ep) = Σ wᵢ · scoreᵢ(ep)` (her scorer kendi içinde 0..1'e normalize eder). + +**İki aşamalı karar.** Heterojen topoloji yüzünden önce havuz, sonra replika seçilir: + +1. **Havuz seçimi:** `Priority` en yüksek havuz; oradaki tüm replikalar doygunsa (§7.2.1) bir + sonraki havuza taşma (spillover). Böylece H100 havuzu dolmadan A100 havuzuna düşülmez. +2. **Replika seçimi:** seçilen havuz içinde filter/score/pick. + +**Yönlendirme profilleri** (config'ten, model başına varsayılan + `x-llmgw-profile` header'ı ile +istek başına override): + +| Profil | Ağırlıklar | Kullanım | +|---|---|---| +| `chat` | prefix 3.0, queue 1.0, kv 1.0 | Çok turlu sohbet, agent — cache tekrarı yüksek | +| `batch` | queue 3.0, kv 1.0, prefix 0.5 | Toplu iş — throughput önemli, TTFT değil | +| `long-context` | kv 3.0, prefix 2.0, queue 1.0 | Uzun prompt — KV baskısı asıl darboğaz | +| `debug` | round-robin picker | Hata ayıklama | + +### 7.4 Scorer'lar ve formülleri + +| Scorer | Formül | Not | +|---|---|---| +| `queue` | `1 − min(1, waiting / target_waiting)` | Kuyruk derinliği | +| `kv` | `1 − kv_cache_usage` | KV doluluğu; tavan üstü sert ceza | +| `prefix` | `matched_blocks / prompt_blocks` | §7.5 | +| `load` | `1 − norm(running / drain_rate)` | AIBrix: `drain_rate` = gözlenen tamamlanma hızı. Yavaş bir replikayı `running` tek başına yakalayamaz. | +| `inflight` | `1 − norm(router_inflight)` | Router'ın kendi saydığı; scrape gecikmesini telafi eder. **Scrape aralığı 2 sn iken bu olmadan sürü etkisi kaçınılmaz.** | +| `pool-priority` | profile `Priority`'den sabit | Havuz tercihi | +| `session` | session hash eşleşmesi | §7.5 Kademe 0 | +| `lora` (v2) | adapter yüklü mü | `vllm:lora_requests_info` | + +**Alternatif maliyet formu (Dynamo).** Prefill ve decode maliyetini doğrudan blok cinsinden +modellemek, normalize skorlardan daha isabetli olabilir: + +``` +cost(ep) = w_overlap · (prompt_blocks − cached_blocks(ep)) + decode_blocks(ep) +seçim = argmin cost +``` + +`w_overlap` yüksek → cache tekrarını kollar (TTFT iyileşir); düşük → yükü eşitler (ITL iyileşir). +Her ikisi de `Scorer` olarak ifade edilebildiği için çerçeve bozulmaz; v1'de ağırlıklı skor, +v1.1'de bu maliyet formu bir scorer olarak eklenir ve `ctl bench` ile karşılaştırılır. + +**Picker'lar:** +- `max-score`: en yüksek skor. Deterministik ama sürü etkisine açık. +- `p2c` (power-of-two-choices): rastgele iki aday, iyisi. Çok replikada varsayılan. +- `softmax(temperature)`: Dynamo'nun `router_temperature`'ı. Skorları logit sayıp örnekler. + `temperature=0` → argmax. Küçük bir sıcaklık, eşit skorlu replikalarda senkronize + kararların yarattığı salınımı kırar. + +### 7.5 Prefix / KV cache farkındalığı — üç kademe + +Bu, router'ın sağlayabileceği en büyük tek kazanç; ama kademeler arasındaki maliyet farkı çok +büyük. Kademeleri ayrı tutuyoruz ki model başına açılıp kapatılabilsin. + +#### Kademe 0 — Session affinity (v1, ucuz, yüksek getiri) +İstemcinin verdiği `x-session-id` header'ı (yoksa OpenAI `user` alanı) bounded-load consistent +hashing ile bir replikaya eşlenir. Replika aşırı yüklüyse (ortalamanın `load_factor` katı) sıradakine +geçilir. Kod olarak ~100 satır; production-stack ölçümlerinde round-robin'e göre cache hit +%93 → %96 civarı. Çok turlu sohbette ve agent iş yüklerinde en iyi maliyet/fayda oranı budur. + +#### Kademe 1 — Yaklaşık radix tree (v1.1, orta maliyet) +Router, gördüğü istek prefix'lerinden **worker'ların cache'ini tahmin eden** bir ağaç tutar +(SGLang router'ın yaklaşımı). Gerçek cache durumunu bilmez; lazy güncellenir, maliyeti düşüktür. + +Eşikler — gerçek implementasyonlarda kritik olan kısım bunlar: +- `cache_threshold`: eşleşme oranı bunun altındaysa "cache yok" sayılır, yük dengeleme devreye girer. +- `balance_abs_threshold` **ve** `balance_rel_threshold`: `max_load − min_load > abs` **ve** + `max_load / min_load > rel` ise **prefix tamamen yok sayılır** ve en az yüklü replikaya gidilir. + (AIBrix de aynı korumayı koyuyor.) Bu koruma olmadan prefix affinity tek bir replikayı boğar — + cache hit oranı artarken p99 gecikme çöker. +- `eviction_interval` + `max_tree_size`: ağaç sınırsız büyümesin. + +Kararın hangi dalından geçtiği metriğe yazılmalı: `cache_hit | cache_miss_min_load | +load_balance_override | stale | no_tree | no_healthy_worker`. Bu olmadan prefix routing'i +ayarlamak körlemesine olur. + +#### Kademe 2 — Kesin KV index (v2, pahalı, opsiyonel) +vLLM `--kv-events-config` ile blok olaylarını ZMQ üzerinden yayınlar (`BlockStored`, +`BlockRemoved`). Router bunlara abone olup `block_hash → {replika}` haritası tutar; gerçek, +küresel bir görünüm elde eder. + +**Tuzak — planın kabul etmesi gereken maliyet:** blok hash'leri vLLM'in **token-id tabanlı** +şemasıyla birebir aynı hesaplanmalıdır. Bu, router'ın istekleri **engine ile aynı tokenizer'la** +tokenize etmesini zorunlu kılar. Go tarafında bu demek: her model için tokenizer yüklemek +(cgo ile HF `tokenizers` veya saf Go bir implementasyon), sürüm uyumunu takip etmek ve istek +başına tokenizasyon CPU maliyetini ödemek. AIBrix bu yüzden "remote tokenizer" zorunluluğu +koyuyor. **Kademe 0 ve 1 tokenizer gerektirmez** — bu yüzden onlar v1'de, bu v2'de. + +Olay kaybı: olay sıra numaralarında boşluk görülürse replay endpoint'inden eksik aralık istenir; +alınamazsa o replikanın index'i düşürülür ve replika **Kademe 1'e** düşer (sessizce yanlış +yönlendirme yapmaktansa yaklaşık bilgiye dönmek doğrudur). Replika yeniden başlarsa index sıfırlanır. + +### 7.6 Engine metrik kontratı + +Bu, "yeni bir engine eklemek" demenin resmî tanımıdır. Bir engine bu üçlüyü Prometheus +formatında verirse llmgw onu yönlendirebilir (GIE'nin model server protokolü): + +| Kontrat alanı | vLLM metriği | Zorunlu | +|---|---|---| +| `Waiting` | `vllm:num_requests_waiting` | ✅ | +| `Running` | `vllm:num_requests_running` | ✅ | +| `KVCacheUsage` | `vllm:kv_cache_usage_perc` | ✅ | +| `BlockSize`, `NumGPUBlocks` | `vllm:cache_config_info` | ➖ | +| LoRA durumu | `vllm:lora_requests_info` (`max_lora`, `running_lora_adapters`, `waiting_lora_adapters`) | ➖ | + +> Düzeltme: v0.1'de `vllm:gpu_cache_usage_perc` yazılmıştı; doğrusu `vllm:kv_cache_usage_perc`. +> Metrik adları sürümle değişiyor — `ParseMetrics` eşlemesi config'ten ezilebilir olmalı ve +> agent, engine sürümünü register ederken bildirmeli. + +Scrape kuralları: +- Her Ready endpoint 1–2 sn'de bir, endpoint başına goroutine, jitter ile. +- Başarısız scrape → endpoint "stale"; skoru kötüleşir ama hemen elenmez. +- 3 ardışık hata → breaker (§7.7). +- **Stale veriyle karar vermemek için** `inflight` scorer'ı (§7.4) her zaman açık kalır. + +### 7.7 Güvenilirlik + +- **Retry:** sadece upstream'e ilk byte yazılmadan önce ve bağlantı hatası / 503 / 429 durumunda, + farklı endpoint ile, en fazla N kez. Stream başladıktan sonra retry yok. +- **Cooldown (LiteLLM dersi):** hata veren endpoint `cooldown_seconds` boyunca aday havuzundan + çıkar. Uzun bir fallback listesi + cooldown'suz retry = retry fırtınası; bir endpoint'in + rate limit'i tüm kümeye yayılır. +- **Circuit breaker:** endpoint başına kapalı/açık/yarı-açık durum makinesi, üstel backoff. +- **Fallback:** modelin tüm endpoint'leri yok/dolu → `Fallbacks` listesindeki modele (key izin + veriyorsa). Yanıta `x-llmgw-served-model` header'ı. **Bütçe fallback'i:** key bütçesini aştıysa + reddetmek yerine daha ucuz bir modele düşürme seçeneği (§7.9). +- **Timeout:** bağlantı, ilk token (TTFT) ve toplam için ayrı; streaming'de token arası idle timeout. +- **İptal:** istemci bağlantıyı koparırsa upstream isteği de iptal et (context propagation) — + GPU boşa çalışmasın. Kuyrukta bekleyen istekler için de geçerli. +- **Drain:** replika `Draining` iken yeni istek almaz; aktif istekler bitene veya timeout'a kadar bekler. +- **Control plane kopması:** router son bilinen endpoint listesiyle çalışmaya devam eder (§14). + +### 7.8 Streaming proxy detayları + +- `http.Transport`: yüksek `MaxIdleConnsPerHost`, HTTP/1.1 keep-alive (engine'ler genelde h2c + desteklemez). +- `bufio.Reader` ile satır oku, `data:` satırlarını geçir, her event sonrası `http.Flusher.Flush()`. +- Usage için istemci `stream_options.include_usage` göndermemişse router ekleyebilir (config ile), + son chunk'taki usage'ı yakalar; istemci istememişse o chunk'ı istemciye iletmez. +- Upstream'in hata JSON'larını OpenAI formatına normalize et. +- Yanıt header'ları: `x-llmgw-replica`, `x-llmgw-pool`, `x-llmgw-served-model`, + `x-llmgw-queue-ms` (admission'da beklenen süre), `x-llmgw-cache-decision` (§7.5). + +### 7.9 Kimlik, kota, kullanım (tenancy) + +LiteLLM'in üretimde kanıtlanmış modeli: + +``` +Organizasyon ─► Takım ─► Proje ─► API Key +``` +Her seviyede: bütçe (token veya sanal maliyet birimi), RPM/TPM limiti, izinli model listesi. +Key'de ayrıca: **öncelik bandı** (§7.2.2) ve **varsayılan yönlendirme profili** (§7.3). + +- API key: `sk-llmgw-...`, DB'de sadece hash (SHA-256). Sızarsa upstream'e dokunmadan döndürülür. +- Rate limit: RPM (`x/time/rate`) + **TPM**. Token limiti istek başında `max_tokens + tahmini prompt` + ile rezerve edilir, yanıt bitince gerçek değerle düzeltilir. (Prompt tahmini v1'de bayt/4; + Kademe 2 tokenizer geldiğinde kesinleşir.) +- **Bütçe:** dönemsel (günlük/aylık) harcama. Aşımda: `429` veya bütçe fallback'i. Maliyeti sıfır + olan modeller bütçe kontrolünden muaf — böylece bütçesi bitmiş bir key'e "ucuz model" + fallback'i tanımlanabilir. +- **Spend yazımı istek yolunun dışında:** usage kaydı kanala → batch halinde SQLite'a; saatlik + rollup. İstek yolunda hiçbir DB yazımı yok. +- **Replika sabitleme (debug):** `x-llmgw-replica` header'ı, key'e scope'lu. Sabitlenen replika + cooldown'da veya yoksa normal yönlendirmeye düşer. +- v1 tek instance → bellek içi limiter. Çok instance olursa Redis veya tutarlı hash ile sticky (v2). + +--- + +## 8. Gözlemlenebilirlik + +Prometheus metrikleri (örnek): +``` +llmgw_requests_total{model,pool,status} +llmgw_request_duration_seconds{model,pool} +llmgw_ttft_seconds{model,pool} +llmgw_tokens_total{model,direction="prompt|completion"} +llmgw_endpoint_inflight{replica} +llmgw_endpoint_score{replica} +llmgw_replicas{model,state} +llmgw_gpu_free{node,gpu_class,fabric} +llmgw_schedule_failures_total{model,reason} +llmgw_agent_connected{node} + +# admission / flow control (§7.2) +llmgw_admission_queue_depth{band,model} +llmgw_admission_wait_seconds{band,model} # histogram +llmgw_admission_rejected_total{band,reason="full|ttl|client_gone"} +llmgw_pool_saturation{pool} # 0..1, dedektör çıktısı + +# yönlendirme kararları (§7.5) — bunlar olmadan prefix routing ayarlanamaz +llmgw_cache_decision_total{model,decision="hit|miss_min_load|balance_override|stale|no_tree"} +llmgw_prefix_match_ratio{model} # histogram +llmgw_scorer_value{scorer,replica} +llmgw_endpoint_metrics_stale{replica} +llmgw_capacity_profile_stale{model,profile} +``` +- OpenTelemetry trace: router span → upstream span (traceparent header engine'e iletilir). +- Yapısal log (slog JSON), `request_id` her yerde. +- `llmgw ctl top`: havuz bazında anlık kuyruk ve doluluk. + +--- + +## 9. Konfigürasyon Örnekleri + +### 9.1 server.yaml +```yaml +listen: ":8080" +admin_listen: ":9090" # metrics, admin API, pprof +agent_listen: ":7443" +tls: + ca: /etc/llmgw/ca.pem + cert: /etc/llmgw/server.pem + key: /etc/llmgw/server-key.pem +store: + path: /var/lib/llmgw/llmgw.db + +fabrics: + - name: ib-a + gid_prefix: "fec0:0000:0000:000a" # opsiyonel otomatik eşleme + - name: ib-b + gid_prefix: "fec0:0000:0000:000b" + +router: + scrape_interval: 2s + inject_usage: true + retries: 2 + cooldown_seconds: 15 # hata veren endpoint bu süre aday havuzdan çıkar + endpoint_cache_max_age: 5m # control plane düşerse bu kadar eskiye kadar çalışmaya devam + + # §7.2 admission — doygunluk dedektörü + admission: + target_waiting_per_replica: 4 # ctl bench önerir + kv_ceiling: 0.90 + queue_ttl: 30s + queue_capacity_per_band: 2000 + bands: + - { name: platinum, priority: 400, saturation_ceiling: 1.00 } + - { name: gold, priority: 300, saturation_ceiling: 0.85 } + - { name: silver, priority: 200, saturation_ceiling: 0.60 } + - { name: bronze, priority: 100, saturation_ceiling: 0.30 } + + # §7.3 yönlendirme profilleri + default_profile: chat + profiles: + chat: + scorers: { prefix: 3.0, queue: 1.0, kv: 1.0, inflight: 1.0 } + picker: { type: p2c } + batch: + scorers: { queue: 3.0, kv: 1.0, load: 1.0, prefix: 0.5 } + picker: { type: max-score } + long-context: + scorers: { kv: 3.0, prefix: 2.0, queue: 1.0 } + picker: { type: softmax, temperature: 0.2 } + + # §7.5 prefix farkındalığı + prefix: + session_affinity: { enabled: true, header: x-session-id, load_factor: 1.25 } # Kademe 0 + approx_tree: # Kademe 1 + enabled: false # v1.1 + cache_threshold: 0.30 + balance_abs_threshold: 16 + balance_rel_threshold: 1.5 + max_tree_size: 2000000 + eviction_interval: 60s + kv_events: { enabled: false } # Kademe 2 — tokenizer gerektirir, v2 + +models: + - name: llama-3.3-70b + source: /models/Llama-3.3-70B-Instruct + scaling: { min: 2, max: 6 } + fallbacks: [qwen2.5-32b] + routing_profile: chat + slo: { ttft_p99: 2s, tpot_p99: 50ms } # ctl bench bu SLO'ya göre kapasite ölçer + profiles: + - name: h100-tp4-fp8 + gpu_class: h100-80g + engine: vllm + tp: 4 + nodes: 1 + priority: 100 + weight: 100 + args: ["--quantization", "fp8", "--max-model-len", "32768"] + - name: a100-tp8 + gpu_class: a100-80g + engine: vllm + tp: 8 + nodes: 1 + priority: 50 + weight: 45 + + - name: deepseek-large + source: /models/DeepSeek-V3 + scaling: { min: 1, max: 2 } + profiles: + - name: h100-2node + gpu_class: h100-80g + engine: sglang + tp: 16 + nodes: 2 # → aynı fabric zorunlu + gpus_per_node: 8 + priority: 100 + weight: 100 + startup_timeout: 40m +``` + +### 9.2 agent.yaml +```yaml +node_id: gpu-a-01 +server: "llmgw-server.internal:7443" +mgmt_interface: eth0 +tls: { ca: ..., cert: ..., key: ... } +labels: { rack: r12 } +fabrics: + - name: ib-a + hcas: [mlx5_0, mlx5_1, mlx5_2, mlx5_3] +port_range: [20000, 20999] # engine HTTP + dist init portları +engines: + vllm: { bin: /opt/venvs/vllm/bin/vllm, launch_mode: ray } + sglang: { python: /opt/venvs/sglang/bin/python } +work_dir: /var/lib/llmgw-agent +``` + +--- + +## 10. Benchmark ve Ağırlıklandırma + +`llmgw ctl bench --profile

`: +- Hedef replikaya sabit prompt/çıktı uzunluklarıyla artan eşzamanlılıkta yük uygular. +- TTFT p50/p95, TPOT, throughput (tok/s), SLO'yu karşılayan maksimum eşzamanlılık ölçülür. +- Çıktı §6.2.1'deki `CapacityProfile`'dır: `MaxConcurrencyAtSLO`, throughput, TTFT/TPOT p99. +- `--apply` ile üç değeri birden günceller: profile `weight`, `target_waiting_per_replica`, + autoscaling hedefi. +- Farklı GPU sınıflarını karşılaştırmanın tek sağlıklı yolu bu; tahmini ağırlık kullanma. +- `--sweep-prefix` modu: prefix cache hit oranını 0/50/90% senaryolarıyla ölçer — yönlendirme + profillerinin ağırlıklarını (§7.3) körlemesine değil ölçerek ayarlamak için. +- Engine sürümü değişince profil bayatlar; `ctl bench` sonucu engine sürümüyle etiketlenir. + +--- + +## 11. Yol Haritası + +### Faz 0 — İskelet (1 hafta) +- [ ] `go mod init`, cobra komutları, koanf config, slog, Makefile, golangci-lint +- [ ] `sim-engine`: OpenAI uyumlu sahte engine (ayarlanabilir gecikme, token hızı, kuyruk, `/metrics` vLLM formatında) +- [ ] CI: `go test ./...`, race detector + +### Faz 1 — Statik router (2 hafta) +- [ ] OpenAI handler'ları, streaming proxy, iptal propagasyonu +- [ ] Config'ten statik endpoint listesi +- [ ] **`Filter`/`Scorer`/`Picker` çerçevesi (§7.3)** — tek bir "strateji" değil, en baştan eklenti +- [ ] `queue`, `kv`, `inflight` scorer'ları; `max-score` ve `p2c` picker'ları +- [ ] Session affinity (§7.5 Kademe 0) — ucuz, erken kazanç +- [ ] Metrik scrape (§7.6 kontratı), stale tespiti, retry, cooldown, breaker +- [ ] Prometheus metrikleri + karar metrikleri (§8) +- **Çıkış kriteri:** 3 sim-engine arkasında 1000 eşzamanlı stream; istemci kopmalarında upstream'in + iptal edildiği doğrulanmış; router overhead p99 < 2 ms. + +### Faz 1.5 — Admission control (1 hafta) ⭐ +- [ ] Doygunluk dedektörü (§7.2.1) +- [ ] Öncelik bantları, bant içi tenant round-robin, kuyruk kapasitesi + TTL +- [ ] `Admission` interface + lease, kuyrukta iptal +- [ ] `llmgw_admission_*` metrikleri +- **Çıkış kriteri:** bronze banttan 500 eşzamanlı batch isteği akarken platinum bandından gelen + interaktif isteklerin TTFT p99'u boştaki değerin 1.5 katını aşmıyor. **Bu testi yazmadan + ileri gitme — Faz 1.5'in tek varlık sebebi bu.** + +### Faz 2 — Agent: keşif + tek node engine (2 hafta) +- [ ] GPU keşfi (NVML + fallback), IB keşfi, topoloji +- [ ] Engine interface, vLLM ve SGLang tek-node +- [ ] Supervisor (process group, log tamponu, durum makinesi, sahiplenme) +- [ ] connect-go stream, register/heartbeat, mTLS +- **Çıkış kriteri:** gerçek bir node'da `ctl` ile model başlat/durdur. + +### Faz 3 — Control plane (2–3 hafta) +- [ ] SQLite store + migration +- [ ] Katalog, envanter, scheduler (tek-node profile'lar), reconciler +- [ ] Router'ın `EndpointSource`'u dinamik hale gelir +- [ ] `ctl`: nodes, models, replicas, cordon, drain, explain-placement, logs +- **Çıkış kriteri:** node kapatılınca replikalar başka node'a taşınıyor; router kesintisiz yönlendiriyor. + +### Faz 4 — Multi-node & fabric (2 hafta) +- [ ] Fabric kimliği (config + GID prefix), doğrulama +- [ ] Multi-node placement (fabric kısıtı), port ayırma, rank sıralı başlatma +- [ ] NCCL env üretimi, HCA seçimi +- [ ] Replika bazlı hata yönetimi (bir shard ölürse tümü) +- **Çıkış kriteri:** her iki fabric'te 2-node replika; bir node'u öldürünce replika aynı fabric'te yeniden kuruluyor, asla diğer fabric'e yayılmıyor. + +### Faz 5 — Üretim özellikleri (2–3 hafta) +- [ ] Tenancy: org/takım/proje/key hiyerarşisi, bütçe, RPM/TPM, usage kaydı (§7.9) +- [ ] Bütçe fallback'i, model allowlist, key→band eşlemesi +- [ ] Yönlendirme profilleri config'ten + `x-llmgw-profile` header'ı +- [ ] Yaklaşık radix tree (§7.5 Kademe 1) — eşikler ve karar metrikleriyle birlikte +- [ ] Fallback modeller, havuz taşması (spillover) +- [ ] `ctl bench` → `CapacityProfile` (§6.2.1), `ctl plan` +- [ ] OTel tracing +- [ ] systemd unit'leri, dokümantasyon + +### Faz 6 — v2 fikirleri +- [ ] Kesin KV index (§7.5 Kademe 2) — vLLM KV event'leri + tokenizer entegrasyonu +- [ ] Dynamo tarzı blok maliyet fonksiyonu, `softmax` picker'ın ayarlanması +- [ ] Admission displacement (düşük banttan istek düşürme) +- [ ] Autoscaling ve öncelik bazlı preemption +- [ ] Control plane HA (Raft) +- [ ] Ağırlık indirme/cache yönetimi +- [ ] Prefill/decode ayrıştırması (§12) +- [ ] LoRA adapter yönlendirmesi (adapter'ı yüklü replikaya affinity) +- [ ] Dış provider desteği (fallback olarak) +- [ ] Admin web arayüzü + +--- + +## 12. Prefill/Decode Ayrıştırması (v2 notları) +- Prefill ve decode replikaları **aynı fabric** içinde olmalı (KV transferi RDMA üzerinden). +- Router'da iki aşamalı akış: prefill replikasına gönder → KV handle → decode replikası. +- Engine desteği sürümlere çok bağlı (vLLM KV connector'ları, SGLang PD modu, NVIDIA Dynamo/NIXL). Kendi implementasyon yerine önce engine'in yerleşik PD router'ını bir "endpoint" gibi sarmak daha az riskli. +- llm-d ve Dynamo'nun gösterdiği: PD'nin kazancı esas olarak **uzun context**'te. Kısa promptlu + sohbet trafiğinde KV transferi maliyeti kazancı yiyor. Önce §10'daki bench ile iş yükünüzün + ISL/OSL dağılımını ölçün; PD'yi ancak uzun-context modelleri için açın. +- Scheduler'a `Role: prefill|decode` ve "aynı fabric'te eşli yerleşim" kısıtı eklenir. + +--- + +## 13. Test Stratejisi +- **Birim:** scheduler (tablo tabanlı: heterojen envanter + beklenen yerleşim / hata sebebi), balancer skorları, SSE parse, metrik parse (gerçek vLLM/SGLang `/metrics` çıktılarından fixture). +- **Özellik tabanlı:** scheduler çıktısında "multi-node replika tek fabric'te" ve "tek GPU class" invariant'ları her zaman doğru (`testing/quick` veya `rapid`). +- **Entegrasyon:** tek süreçte server + N sahte agent (sahte GPU/IB envanteri) + sim-engine'ler. +- **Kaos:** agent öldürme, ağ kesintisi (toxiproxy), engine çökmesi, yavaş engine, control plane'i + öldürüp router'ın önbellekle devam ettiğini doğrulama. +- **Öncelik izolasyonu:** Faz 1.5 çıkış kriteri — düşük öncelikli yük altında yüksek öncelikli + TTFT'nin korunduğu regresyon testi. Bu test CI'da kalıcı olmalı. +- **Yönlendirme kalitesi:** sim-engine'e gerçekçi bir prefix cache modeli koy (blok bazlı LRU), + sentetik çok-turlu trafikle cache hit oranını ölç. round-robin / session-affinity / approx-tree + karşılaştırması sayısal olsun. +- **Yük:** `k6` veya kendi Go yük üreticimiz; router'ın kendi overhead'ini ölç (hedef: p99 < 2 ms ek gecikme). +- **Gerçek donanım:** her fazın sonunda en az bir IB'li ve bir IB'siz node'da smoke test. + +--- + +## 14. Riskler ve Açık Sorular +| Konu | Not | +|---|---| +| Engine sürüm farkları | CLI bayrakları ve metrik adları sık değişiyor → engine adapter'larını sürüm eşlemeli ve config'le ezilebilir yap. | +| vLLM multi-node başlatma | Ray mı yerleşik mp mi? Kullanılacak sürümde test edilmeli. | +| Fabric tespiti | OpenSM'lerde benzersiz subnet prefix ayarlanabilir mi? Değilse manuel etiket şart. | +| Aynı node'da çok fabric | Var mı? Varsa HCA-GPU NUMA eşleşmesi önemli. | +| Model depolama | Paylaşımlı FS var mı, yükleme süreleri? Startup timeout'ları buna göre. | +| Karma sürücü/CUDA sürümleri | Engine venv'leri GPU sınıfına göre farklı olmalı mı? (`engines` config'i profile başına override edilebilmeli.) | +| Tek control plane | v1'de SPOF. Router endpoint listesini önbellekte tutar (§6.4), control plane kısa süre düşse de çalışmaya devam eder. | +| Güvenlik | Engine portları sadece router'a açık olmalı (firewall / bind adresi). | +| **Tokenizer tutarlılığı** | Kesin KV index (§7.5 Kademe 2) router'ın engine ile **aynı tokenizer**'ı kullanmasını zorunlu kılar. Go'da bu cgo veya ayrı bir tokenizer servisi demek. Bu yüzden v2'ye bırakıldı; v1'in prefix çözümü tokenizer gerektirmiyor. TPM limitindeki prompt tahmini de bu yüzden v1'de yaklaşık (bayt/4). | +| **Prefix affinity'nin ters tepmesi** | Yük dengesizliği koruması (§7.5 Kademe 1 eşikleri) olmadan prefix routing cache hit oranını artırırken tek replikayı boğar ve p99'u bozar. Eşikler açılmadan prefix affinity **açılmamalı**. | +| **Admission eşiklerinin ayarı** | `target_waiting_per_replica` çok düşükse GPU aç kalır, çok yüksekse kuyruk gecikmesi geri gelir. Bu tek sayı sistemin davranışını belirliyor; `ctl bench` ile ölçülmeli, tahmin edilmemeli. | +| KV event kaybı | ZMQ fire-and-forget; olay kaybında router sessizce yanlış yönlendirir. Sıra numarası boşluğu tespiti + Kademe 1'e düşme zorunlu (§7.5). | +| Metrik bayatlığı | 2 sn'lik scrape aralığında 1000 istek/sn'de karar tamamen bayat veriye dayanır. `inflight` scorer'ı bu yüzden opsiyonel değil. | + +--- + +## 15. İlk Adımlar (VS Code'da buradan devam) +1. `mkdir llmgw && cd llmgw && go mod init /llmgw` +2. `cmd/llmgw/main.go`: cobra ile `server`, `agent`, `ctl`, `sim-engine` alt komutlarının boş iskeleti. +3. `internal/sim`: sahte engine — ilk yazılacak gerçek kod bu olsun, geri kalan her şey buna karşı test edilecek. +4. `internal/router/proxy`: streaming proxy + iptal testi. +5. `internal/router/scheduler`: `Filter`/`Scorer`/`Picker` interface'leri + `queue`, `kv`, + `inflight` scorer'ları. **Tek bir "least-load" fonksiyonu yazma** — çerçeveyi en baştan kur, + sonradan geri dönmek pahalı. +6. `internal/router/admission`: doygunluk dedektörü + bant kuyrukları. Faz 1.5. +7. Faz 1 ve Faz 1.5 çıkış kriterlerini sağlayan e2e testlerini yaz, sonra Faz 2'ye geç. + +> `internal/router/balancer/` yerine `internal/router/scheduler/` (§4'teki ağaç buna göre +> güncellenmeli): "balancer" ismi yaptığı işten dar kalıyor — bu bileşen yük dengelemekten +> fazlasını yapıyor.