1044 lines
34 KiB
Go
1044 lines
34 KiB
Go
package loadharness
|
|
|
|
import (
|
|
"context"
|
|
"crypto/md5"
|
|
cryptorand "crypto/rand"
|
|
"crypto/rsa"
|
|
"encoding/binary"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"net"
|
|
"os"
|
|
"runtime"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/iamxvbaba/td/mtproto"
|
|
"github.com/iamxvbaba/td/pool"
|
|
tdrpc "github.com/iamxvbaba/td/rpc"
|
|
"github.com/iamxvbaba/td/telegram"
|
|
"github.com/iamxvbaba/td/tg"
|
|
)
|
|
|
|
const (
|
|
workerConnecting int32 = iota
|
|
workerReady
|
|
workerDisconnected
|
|
workerOffline
|
|
workerStopped
|
|
)
|
|
|
|
type RunConfig struct {
|
|
ManifestPath string
|
|
SessionKeyPath string
|
|
RSAKeyOverride string
|
|
ReportPath string
|
|
EventsPath string
|
|
FileFixturePath string
|
|
ServerMetricsURL string
|
|
SessionLimit int
|
|
Duration time.Duration
|
|
RecoveryDuration time.Duration
|
|
RampDuration time.Duration
|
|
RPCInterval time.Duration
|
|
MessageInterval time.Duration
|
|
FileInterval time.Duration
|
|
FileSizeBytes int
|
|
FileChunkBytes int
|
|
SetupTimeout time.Duration
|
|
OperationTimeout time.Duration
|
|
SampleInterval time.Duration
|
|
OfflineFraction float64
|
|
OfflineAt time.Duration
|
|
OfflineFor time.Duration
|
|
MinimumReadyRatio float64
|
|
ExpectServerRestart bool
|
|
}
|
|
|
|
func (c RunConfig) validate() error {
|
|
if c.ManifestPath == "" || c.SessionKeyPath == "" || c.ReportPath == "" {
|
|
return errors.New("manifest, session-key and report paths are required")
|
|
}
|
|
if c.Duration <= 0 || c.RecoveryDuration < 0 || c.RampDuration < 0 || c.RPCInterval <= 0 || c.OperationTimeout <= 0 || c.SampleInterval <= 0 {
|
|
return errors.New("run durations and intervals are invalid")
|
|
}
|
|
if c.FileSizeBytes < 0 || c.FileChunkBytes < 0 || c.FileChunkBytes > 1<<20 || c.FileSizeBytes > 64<<20 {
|
|
return errors.New("file size must be <=64MiB and chunk size must be <=1MiB")
|
|
}
|
|
if c.FileSizeBytes > 0 && (c.FileInterval <= 0 || c.FileChunkBytes <= 0) {
|
|
return errors.New("enabled file workload requires positive interval and chunk size")
|
|
}
|
|
if c.FileSizeBytes > 0 && c.SetupTimeout <= 0 {
|
|
return errors.New("enabled file workload requires a positive setup timeout")
|
|
}
|
|
if c.OfflineFraction < 0 || c.OfflineFraction > 1 || c.MinimumReadyRatio <= 0 || c.MinimumReadyRatio > 1 {
|
|
return errors.New("offline fraction and minimum ready ratio must be within [0,1]")
|
|
}
|
|
if c.OfflineFraction > 0 && (c.OfflineAt <= 0 || c.OfflineFor <= 0 || c.OfflineAt+c.OfflineFor >= c.Duration) {
|
|
return errors.New("offline window must be positive and fit inside load duration")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type updateState struct {
|
|
mu sync.Mutex
|
|
value tg.UpdatesState
|
|
valid bool
|
|
}
|
|
|
|
func (s *updateState) load() (tg.UpdatesState, bool) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.value, s.valid
|
|
}
|
|
|
|
func (s *updateState) store(value tg.UpdatesState) {
|
|
s.mu.Lock()
|
|
s.value = value
|
|
s.valid = true
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
type harnessCounters struct {
|
|
connectionAttempts atomic.Uint64
|
|
reconnects atomic.Uint64
|
|
disconnects atomic.Uint64
|
|
updates atomic.Uint64
|
|
fatalErrors atomic.Uint64
|
|
downloadBytes atomic.Uint64
|
|
}
|
|
|
|
var debugConnectionErrors atomic.Uint64
|
|
var debugOperationErrors atomic.Uint64
|
|
|
|
func debugConnectionError(sessionIndex int, err error) {
|
|
if os.Getenv("TELESRV_LOAD_DEBUG_ERRORS") != "1" || err == nil || debugConnectionErrors.Add(1) > 20 {
|
|
return
|
|
}
|
|
// Explicit opt-in diagnostics go only to stderr and are bounded. They are
|
|
// never copied into the attachable report or event stream.
|
|
fmt.Fprintf(os.Stderr, "load debug: session=%d error_type=%T error=%v\n", sessionIndex, err, err)
|
|
}
|
|
|
|
func debugOperationError(operation string, err error) {
|
|
// Connection lifecycle errors have their own bounded diagnostic path. Do not
|
|
// let an expected reconnect storm consume the operation-error budget that is
|
|
// needed to diagnose actual RPC failures.
|
|
if operation == "connection.dead" || os.Getenv("TELESRV_LOAD_DEBUG_ERRORS") != "1" || err == nil || debugOperationErrors.Add(1) > 20 {
|
|
return
|
|
}
|
|
// Operation names are a code-owned finite vocabulary. Raw errors are
|
|
// opt-in, stderr-only and bounded; reports/events retain only classifications.
|
|
fmt.Fprintf(os.Stderr, "load debug: operation=%s error_type=%T error=%v\n", operation, err, err)
|
|
}
|
|
|
|
type downloadFixture struct {
|
|
location *tg.InputDocumentFileLocation
|
|
size int
|
|
chunk int
|
|
}
|
|
|
|
type loadWorker struct {
|
|
record SessionRecord
|
|
target SessionRecord
|
|
endpoint Endpoint
|
|
publicKey *rsa.PublicKey
|
|
storage *EncryptedFileStorage
|
|
metrics *metricSet
|
|
counters *harnessCounters
|
|
events *eventWriter
|
|
rpcInterval time.Duration
|
|
msgInterval time.Duration
|
|
fileInterval time.Duration
|
|
operationTimeout time.Duration
|
|
fileFixture *downloadFixture
|
|
|
|
desired atomic.Bool
|
|
state atomic.Int32
|
|
everReady atomic.Bool
|
|
signal chan struct{}
|
|
lastUpdate updateState
|
|
messageSeq atomic.Uint64
|
|
}
|
|
|
|
func newLoadWorker(record, target SessionRecord, endpoint Endpoint, publicKey *rsa.PublicKey, storage *EncryptedFileStorage, metrics *metricSet, counters *harnessCounters, events *eventWriter, rpcInterval, messageInterval, fileInterval, operationTimeout time.Duration, fixture *downloadFixture) *loadWorker {
|
|
w := &loadWorker{
|
|
record: record, target: target, endpoint: endpoint, publicKey: publicKey, storage: storage,
|
|
metrics: metrics, counters: counters, events: events, rpcInterval: rpcInterval,
|
|
msgInterval: messageInterval, fileInterval: fileInterval, operationTimeout: operationTimeout, fileFixture: fixture,
|
|
signal: make(chan struct{}, 1),
|
|
}
|
|
w.state.Store(workerStopped)
|
|
return w
|
|
}
|
|
|
|
func (w *loadWorker) operationContext(parent context.Context) (context.Context, context.CancelFunc) {
|
|
return context.WithTimeout(parent, w.operationTimeout)
|
|
}
|
|
|
|
func (w *loadWorker) setOnline(online bool) {
|
|
w.desired.Store(online)
|
|
select {
|
|
case w.signal <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (w *loadWorker) supervise(ctx context.Context, wg *sync.WaitGroup) {
|
|
defer wg.Done()
|
|
for {
|
|
if err := ctx.Err(); err != nil {
|
|
w.state.Store(workerStopped)
|
|
return
|
|
}
|
|
if !w.desired.Load() {
|
|
w.state.Store(workerOffline)
|
|
select {
|
|
case <-ctx.Done():
|
|
w.state.Store(workerStopped)
|
|
return
|
|
case <-w.signal:
|
|
continue
|
|
}
|
|
}
|
|
|
|
clientCtx, cancel := context.WithCancel(ctx)
|
|
done := make(chan error, 1)
|
|
go func() { done <- w.runClient(clientCtx) }()
|
|
for w.desired.Load() {
|
|
select {
|
|
case <-ctx.Done():
|
|
cancel()
|
|
<-done
|
|
w.state.Store(workerStopped)
|
|
return
|
|
case <-w.signal:
|
|
if !w.desired.Load() {
|
|
cancel()
|
|
<-done
|
|
w.state.Store(workerOffline)
|
|
}
|
|
case err := <-done:
|
|
cancel()
|
|
if ctx.Err() != nil {
|
|
w.state.Store(workerStopped)
|
|
return
|
|
}
|
|
if !w.desired.Load() {
|
|
w.state.Store(workerOffline)
|
|
goto nextClient
|
|
}
|
|
w.counters.fatalErrors.Add(1)
|
|
w.events.write(map[string]any{
|
|
"type": "worker_error", "at": time.Now().UTC(), "session_index": w.record.Index,
|
|
"class": classifyError(err),
|
|
})
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(time.Second):
|
|
}
|
|
goto nextClient
|
|
}
|
|
}
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
default:
|
|
}
|
|
nextClient:
|
|
}
|
|
}
|
|
|
|
func (w *loadWorker) runClient(ctx context.Context) error {
|
|
reconnectSignal := make(chan struct{}, 1)
|
|
client, err := newClient(w.endpoint, w.publicKey, w.storage, clientHooks{
|
|
Update: telegram.UpdateHandlerFunc(func(context.Context, tg.UpdatesClass) error {
|
|
w.counters.updates.Add(1)
|
|
return nil
|
|
}),
|
|
ConnectionState: func(state telegram.ConnectionState) {
|
|
switch state {
|
|
case telegram.ConnectionStateConnecting:
|
|
w.state.Store(workerConnecting)
|
|
w.counters.connectionAttempts.Add(1)
|
|
if w.everReady.Load() {
|
|
w.counters.reconnects.Add(1)
|
|
}
|
|
case telegram.ConnectionStateReady:
|
|
wasReady := w.everReady.Swap(true)
|
|
w.state.Store(workerReady)
|
|
if wasReady {
|
|
select {
|
|
case reconnectSignal <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
case telegram.ConnectionStateDisconnected:
|
|
w.state.Store(workerDisconnected)
|
|
w.counters.disconnects.Add(1)
|
|
}
|
|
},
|
|
Dead: func(err error) {
|
|
debugConnectionError(w.record.Index, err)
|
|
w.metrics.observe("connection.dead", time.Now(), err)
|
|
w.events.write(map[string]any{
|
|
"type": "connection_dead", "at": time.Now().UTC(), "session_index": w.record.Index,
|
|
"class": classifyError(err), "reason": classifyErrorReason(err),
|
|
})
|
|
},
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return client.Run(ctx, func(ctx context.Context) error {
|
|
statusStart := time.Now()
|
|
operationCtx, cancelOperation := w.operationContext(ctx)
|
|
status, err := client.Auth().Status(operationCtx)
|
|
cancelOperation()
|
|
w.metrics.observe("auth.status", statusStart, err)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !status.Authorized || status.User == nil || status.User.ID != w.record.UserID {
|
|
return errors.New("session is not authorized for its manifest user")
|
|
}
|
|
raw := tg.NewClient(client)
|
|
if _, valid := w.lastUpdate.load(); valid {
|
|
w.catchUp(ctx, raw)
|
|
} else {
|
|
w.refreshUpdateState(ctx, raw)
|
|
}
|
|
|
|
rpcTicker := time.NewTicker(w.rpcInterval)
|
|
defer rpcTicker.Stop()
|
|
var messageTicker *time.Ticker
|
|
var messageC <-chan time.Time
|
|
if w.msgInterval > 0 && w.record.DeviceIndex == 0 && w.target.UserID > 0 {
|
|
messageTicker = time.NewTicker(w.msgInterval)
|
|
messageC = messageTicker.C
|
|
defer messageTicker.Stop()
|
|
}
|
|
var fileTicker *time.Ticker
|
|
var fileC <-chan time.Time
|
|
if w.fileFixture != nil && w.fileInterval > 0 {
|
|
fileTicker = time.NewTicker(w.fileInterval)
|
|
fileC = fileTicker.C
|
|
defer fileTicker.Stop()
|
|
}
|
|
cycle := 0
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-reconnectSignal:
|
|
w.catchUp(ctx, raw)
|
|
case <-rpcTicker.C:
|
|
w.runRPC(ctx, client, raw, cycle)
|
|
cycle++
|
|
case <-messageC:
|
|
w.sendMessage(ctx, raw)
|
|
case <-fileC:
|
|
w.downloadFileChunk(ctx, raw)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
func (w *loadWorker) runRPC(ctx context.Context, client *telegram.Client, raw *tg.Client, cycle int) {
|
|
start := time.Now()
|
|
operationCtx, cancel := w.operationContext(ctx)
|
|
defer cancel()
|
|
var err error
|
|
switch cycle % 4 {
|
|
case 0:
|
|
err = client.Ping(operationCtx)
|
|
w.metrics.observe("ping", start, err)
|
|
case 1:
|
|
var state *tg.UpdatesState
|
|
state, err = raw.UpdatesGetState(operationCtx)
|
|
if err == nil {
|
|
w.lastUpdate.store(*state)
|
|
}
|
|
w.metrics.observe("updates.getState", start, err)
|
|
case 2:
|
|
_, err = raw.MessagesGetDialogs(operationCtx, &tg.MessagesGetDialogsRequest{OffsetPeer: &tg.InputPeerEmpty{}, Limit: 20})
|
|
w.metrics.observe("messages.getDialogs", start, err)
|
|
case 3:
|
|
_, err = raw.HelpGetConfig(operationCtx)
|
|
w.metrics.observe("help.getConfig", start, err)
|
|
}
|
|
}
|
|
|
|
func (w *loadWorker) refreshUpdateState(ctx context.Context, raw *tg.Client) {
|
|
start := time.Now()
|
|
operationCtx, cancel := w.operationContext(ctx)
|
|
state, err := raw.UpdatesGetState(operationCtx)
|
|
cancel()
|
|
w.metrics.observe("updates.getState", start, err)
|
|
if err == nil {
|
|
w.lastUpdate.store(*state)
|
|
}
|
|
}
|
|
|
|
func (w *loadWorker) catchUp(ctx context.Context, raw *tg.Client) {
|
|
state, valid := w.lastUpdate.load()
|
|
if !valid {
|
|
w.refreshUpdateState(ctx, raw)
|
|
return
|
|
}
|
|
for page := 0; page < 32; page++ {
|
|
start := time.Now()
|
|
operationCtx, cancel := w.operationContext(ctx)
|
|
difference, err := raw.UpdatesGetDifference(operationCtx, &tg.UpdatesGetDifferenceRequest{Pts: state.Pts, Date: state.Date, Qts: state.Qts})
|
|
cancel()
|
|
w.metrics.observe("updates.getDifference", start, err)
|
|
if err != nil {
|
|
return
|
|
}
|
|
switch value := difference.(type) {
|
|
case *tg.UpdatesDifferenceEmpty:
|
|
state.Date, state.Seq = value.Date, value.Seq
|
|
w.lastUpdate.store(state)
|
|
return
|
|
case *tg.UpdatesDifference:
|
|
state = value.State
|
|
w.counters.updates.Add(uint64(len(value.NewMessages) + len(value.NewEncryptedMessages) + len(value.OtherUpdates)))
|
|
w.lastUpdate.store(state)
|
|
return
|
|
case *tg.UpdatesDifferenceSlice:
|
|
state = value.IntermediateState
|
|
w.counters.updates.Add(uint64(len(value.NewMessages) + len(value.NewEncryptedMessages) + len(value.OtherUpdates)))
|
|
w.lastUpdate.store(state)
|
|
case *tg.UpdatesDifferenceTooLong:
|
|
state.Pts = value.Pts
|
|
w.lastUpdate.store(state)
|
|
return
|
|
default:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (w *loadWorker) sendMessage(ctx context.Context, raw *tg.Client) {
|
|
sequence := w.messageSeq.Add(1)
|
|
var randomBytes [8]byte
|
|
if _, err := cryptorand.Read(randomBytes[:]); err != nil {
|
|
return
|
|
}
|
|
randomID := int64(binary.LittleEndian.Uint64(randomBytes[:]))
|
|
if randomID == 0 {
|
|
randomID = int64(sequence)
|
|
}
|
|
start := time.Now()
|
|
operationCtx, cancel := w.operationContext(ctx)
|
|
_, err := raw.MessagesSendMessage(operationCtx, &tg.MessagesSendMessageRequest{
|
|
Peer: &tg.InputPeerUser{UserID: w.target.UserID, AccessHash: w.target.AccessHash},
|
|
Message: fmt.Sprintf("load/%d/%d", w.record.Index, sequence), RandomID: randomID,
|
|
})
|
|
cancel()
|
|
w.metrics.observe("messages.sendMessage", start, err)
|
|
}
|
|
|
|
func (w *loadWorker) downloadFileChunk(ctx context.Context, raw *tg.Client) {
|
|
fixture := w.fileFixture
|
|
if fixture == nil || fixture.size <= 0 || fixture.chunk <= 0 {
|
|
return
|
|
}
|
|
sequence := w.messageSeq.Add(1)
|
|
offset := int64((int(sequence-1) * fixture.chunk) % fixture.size)
|
|
limit := min(fixture.chunk, fixture.size-int(offset))
|
|
start := time.Now()
|
|
operationCtx, cancel := w.operationContext(ctx)
|
|
result, err := raw.UploadGetFile(operationCtx, &tg.UploadGetFileRequest{
|
|
Location: fixture.location, Offset: offset, Limit: limit,
|
|
})
|
|
cancel()
|
|
if err == nil {
|
|
file, ok := result.(*tg.UploadFile)
|
|
switch {
|
|
case !ok:
|
|
err = fmt.Errorf("upload.getFile returned %T", result)
|
|
case len(file.Bytes) != limit:
|
|
err = fmt.Errorf("upload.getFile bytes=%d want=%d", len(file.Bytes), limit)
|
|
case !validFixtureBytes(file.Bytes, offset):
|
|
err = errors.New("upload.getFile payload mismatch")
|
|
default:
|
|
w.counters.downloadBytes.Add(uint64(len(file.Bytes)))
|
|
}
|
|
}
|
|
w.metrics.observe("upload.getFile", start, err)
|
|
}
|
|
|
|
func prepareDownloadFixture(ctx context.Context, cfg RunConfig, manifest *Manifest, record SessionRecord, key [32]byte, publicKey *rsa.PublicKey, metrics *metricSet) (*downloadFixture, error) {
|
|
data := make([]byte, cfg.FileSizeBytes)
|
|
for i := range data {
|
|
data[i] = fixtureByte(int64(i))
|
|
}
|
|
fileID, err := randomNonZeroInt64()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
storage := &EncryptedFileStorage{Path: resolveSessionPath(cfg.ManifestPath, record), Key: key}
|
|
client, err := newClient(manifest.Endpoint, publicKey, storage, clientHooks{})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var location *tg.InputDocumentFileLocation
|
|
err = client.Run(ctx, func(ctx context.Context) error {
|
|
status, err := client.Auth().Status(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !status.Authorized || status.User == nil || status.User.ID != record.UserID {
|
|
return errors.New("fixture session is not authorized")
|
|
}
|
|
raw := tg.NewClient(client)
|
|
const partSize = 512 << 10
|
|
parts := (len(data) + partSize - 1) / partSize
|
|
big := len(data) > 10<<20
|
|
for part := 0; part < parts; part++ {
|
|
startOffset := part * partSize
|
|
endOffset := min(len(data), startOffset+partSize)
|
|
start := time.Now()
|
|
var saved bool
|
|
if big {
|
|
saved, err = raw.UploadSaveBigFilePart(ctx, &tg.UploadSaveBigFilePartRequest{
|
|
FileID: fileID, FilePart: part, FileTotalParts: parts, Bytes: data[startOffset:endOffset],
|
|
})
|
|
} else {
|
|
saved, err = raw.UploadSaveFilePart(ctx, &tg.UploadSaveFilePartRequest{
|
|
FileID: fileID, FilePart: part, Bytes: data[startOffset:endOffset],
|
|
})
|
|
}
|
|
metrics.observe("upload.saveFilePart", start, err)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !saved {
|
|
return fmt.Errorf("upload part %d was not saved", part)
|
|
}
|
|
}
|
|
var file tg.InputFileClass
|
|
if big {
|
|
file = &tg.InputFileBig{ID: fileID, Parts: parts, Name: "telesrv-load.bin"}
|
|
} else {
|
|
digest := md5.Sum(data)
|
|
file = &tg.InputFile{ID: fileID, Parts: parts, Name: "telesrv-load.bin", MD5Checksum: hex.EncodeToString(digest[:])}
|
|
}
|
|
start := time.Now()
|
|
media, err := raw.MessagesUploadMedia(ctx, &tg.MessagesUploadMediaRequest{
|
|
Peer: &tg.InputPeerEmpty{},
|
|
Media: &tg.InputMediaUploadedDocument{
|
|
File: file, MimeType: "application/octet-stream",
|
|
Attributes: []tg.DocumentAttributeClass{&tg.DocumentAttributeFilename{FileName: "telesrv-load.bin"}},
|
|
},
|
|
})
|
|
metrics.observe("messages.uploadMedia", start, err)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
documentMedia, ok := media.(*tg.MessageMediaDocument)
|
|
if !ok {
|
|
return fmt.Errorf("messages.uploadMedia returned %T", media)
|
|
}
|
|
documentClass, ok := documentMedia.GetDocument()
|
|
if !ok {
|
|
return errors.New("messages.uploadMedia omitted document")
|
|
}
|
|
document, ok := documentClass.(*tg.Document)
|
|
if !ok {
|
|
return fmt.Errorf("uploaded document is %T", documentClass)
|
|
}
|
|
location = &tg.InputDocumentFileLocation{
|
|
ID: document.ID, AccessHash: document.AccessHash,
|
|
FileReference: append([]byte(nil), document.FileReference...),
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if location == nil {
|
|
return nil, errors.New("file fixture completed without a location")
|
|
}
|
|
return &downloadFixture{location: location, size: len(data), chunk: cfg.FileChunkBytes}, nil
|
|
}
|
|
|
|
func loadOrCreateDownloadFixture(ctx context.Context, cfg RunConfig, manifest *Manifest, record SessionRecord, key [32]byte, publicKey *rsa.PublicKey, metrics *metricSet) (*downloadFixture, error) {
|
|
path := resolveFileFixturePath(cfg.ManifestPath, cfg.FileFixturePath)
|
|
fixture, err := loadPersistedFileFixture(path, manifest.Endpoint, cfg.FileSizeBytes, cfg.FileChunkBytes)
|
|
if err == nil {
|
|
return fixture, nil
|
|
}
|
|
if !os.IsNotExist(err) {
|
|
return nil, fmt.Errorf("load file fixture %q: %w", path, err)
|
|
}
|
|
setupCtx, cancel := context.WithTimeout(ctx, cfg.SetupTimeout)
|
|
defer cancel()
|
|
fixture, err = prepareDownloadFixture(setupCtx, cfg, manifest, record, key, publicKey, metrics)
|
|
if err != nil {
|
|
if errors.Is(err, context.DeadlineExceeded) || errors.Is(setupCtx.Err(), context.DeadlineExceeded) {
|
|
return nil, fmt.Errorf("create file fixture timed out after %s", cfg.SetupTimeout)
|
|
}
|
|
return nil, err
|
|
}
|
|
if err := writePersistedFileFixture(path, manifest.Endpoint, fixture); err != nil {
|
|
return nil, fmt.Errorf("persist file fixture: %w", err)
|
|
}
|
|
return fixture, nil
|
|
}
|
|
|
|
func randomNonZeroInt64() (int64, error) {
|
|
var data [8]byte
|
|
if _, err := cryptorand.Read(data[:]); err != nil {
|
|
return 0, err
|
|
}
|
|
value := int64(binary.LittleEndian.Uint64(data[:]) & math.MaxInt64)
|
|
if value == 0 {
|
|
value = 1
|
|
}
|
|
return value, nil
|
|
}
|
|
|
|
func fixtureByte(offset int64) byte {
|
|
return byte((uint64(offset)*31 + 17) % 251)
|
|
}
|
|
|
|
func validFixtureBytes(data []byte, offset int64) bool {
|
|
for i, value := range data {
|
|
if value != fixtureByte(offset+int64(i)) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// Run executes a bounded real-client load and keeps scraping after all workers
|
|
// stop so logical-session/outbox reclamation can be proven rather than assumed.
|
|
func Run(ctx context.Context, cfg RunConfig) (*RunReport, error) {
|
|
if err := cfg.validate(); err != nil {
|
|
return nil, err
|
|
}
|
|
manifest, err := LoadManifest(cfg.ManifestPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
key, err := LoadSessionKey(cfg.SessionKeyPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
publicKey, err := loadManifestPublicKey(cfg.ManifestPath, manifest.Endpoint, cfg.RSAKeyOverride)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
records := manifest.Sessions
|
|
if cfg.SessionLimit > 0 && cfg.SessionLimit < len(records) {
|
|
records = records[:cfg.SessionLimit]
|
|
}
|
|
if len(records) == 0 {
|
|
return nil, errors.New("manifest has no selected sessions")
|
|
}
|
|
if err := validateProcessCapacity(len(records)); err != nil {
|
|
return nil, err
|
|
}
|
|
events, err := newEventWriter(cfg.EventsPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer events.close()
|
|
metrics := newMetricSet("auth.status", "connection.dead", "ping", "updates.getState", "updates.getDifference", "messages.getDialogs", "help.getConfig", "messages.sendMessage", "upload.saveFilePart", "messages.uploadMedia", "upload.getFile")
|
|
counters := &harnessCounters{}
|
|
serverMetrics := newServerMetricsClient(cfg.ServerMetricsURL)
|
|
var baselineServerMetrics map[string]float64
|
|
if serverMetrics != nil {
|
|
if sample, scrapeErr := serverMetrics.scrape(ctx); scrapeErr == nil {
|
|
baselineServerMetrics = sample
|
|
events.write(map[string]any{"type": "server_baseline", "at": time.Now().UTC(), "server_metrics": sample})
|
|
} else {
|
|
events.write(map[string]any{"type": "server_baseline_error", "at": time.Now().UTC(), "class": classifyError(scrapeErr)})
|
|
}
|
|
}
|
|
var fixture *downloadFixture
|
|
if cfg.FileSizeBytes > 0 {
|
|
fixture, err = loadOrCreateDownloadFixture(ctx, cfg, manifest, records[0], key, publicKey, metrics)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("prepare file fixture: %w", err)
|
|
}
|
|
}
|
|
targets := primaryTargets(records)
|
|
workers := make([]*loadWorker, 0, len(records))
|
|
for _, record := range records {
|
|
target := targets[(record.AccountIndex+1)%len(targets)]
|
|
workers = append(workers, newLoadWorker(
|
|
record, target, manifest.Endpoint, publicKey,
|
|
&EncryptedFileStorage{Path: resolveSessionPath(cfg.ManifestPath, record), Key: key},
|
|
metrics, counters, events, cfg.RPCInterval, cfg.MessageInterval, cfg.FileInterval, cfg.OperationTimeout, fixture,
|
|
))
|
|
}
|
|
|
|
startedAt := time.Now().UTC()
|
|
loadCtx, stopLoad := context.WithCancel(ctx)
|
|
var workerWG sync.WaitGroup
|
|
for _, worker := range workers {
|
|
workerWG.Add(1)
|
|
go worker.supervise(loadCtx, &workerWG)
|
|
}
|
|
for i, worker := range workers {
|
|
delay := time.Duration(0)
|
|
if len(workers) > 1 {
|
|
delay = time.Duration(i) * cfg.RampDuration / time.Duration(len(workers)-1)
|
|
}
|
|
go func(w *loadWorker, d time.Duration) {
|
|
timer := time.NewTimer(d)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-loadCtx.Done():
|
|
case <-timer.C:
|
|
w.setOnline(true)
|
|
}
|
|
}(worker, delay)
|
|
}
|
|
|
|
if cfg.OfflineFraction > 0 {
|
|
go runOfflineWindow(loadCtx, workers, cfg.OfflineFraction, cfg.OfflineAt, cfg.OfflineFor, events)
|
|
}
|
|
loadTimer := time.NewTimer(cfg.Duration)
|
|
sampleTicker := time.NewTicker(cfg.SampleInterval)
|
|
peakReady := 0
|
|
steadySamples := 0
|
|
steadyReadySum := 0
|
|
steadyReadyMinimum := len(workers)
|
|
var finalServerMetrics map[string]float64
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
stopLoad()
|
|
workerWG.Wait()
|
|
return nil, ctx.Err()
|
|
case <-sampleTicker.C:
|
|
ready := countWorkerState(workers, workerReady)
|
|
peakReady = max(peakReady, ready)
|
|
elapsed := time.Since(startedAt)
|
|
outsideOffline := cfg.OfflineFraction == 0 || elapsed < cfg.OfflineAt || elapsed > cfg.OfflineAt+cfg.OfflineFor+30*time.Second
|
|
if elapsed >= cfg.RampDuration+30*time.Second && outsideOffline {
|
|
steadySamples++
|
|
steadyReadySum += ready
|
|
steadyReadyMinimum = min(steadyReadyMinimum, ready)
|
|
}
|
|
finalServerMetrics = writeSample(ctx, events, "load", workers, metrics, counters, serverMetrics)
|
|
case <-loadTimer.C:
|
|
goto loadFinished
|
|
}
|
|
}
|
|
|
|
loadFinished:
|
|
sampleTicker.Stop()
|
|
stopLoad()
|
|
workerWG.Wait()
|
|
loadEndedAt := time.Now().UTC()
|
|
if ready := countWorkerState(workers, workerReady); ready > peakReady {
|
|
peakReady = ready
|
|
}
|
|
|
|
if cfg.RecoveryDuration > 0 {
|
|
recoveryDeadline := time.NewTimer(cfg.RecoveryDuration)
|
|
recoveryTicker := time.NewTicker(cfg.SampleInterval)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
recoveryTicker.Stop()
|
|
recoveryDeadline.Stop()
|
|
return nil, ctx.Err()
|
|
case <-recoveryTicker.C:
|
|
finalServerMetrics = writeSample(ctx, events, "recovery", workers, metrics, counters, serverMetrics)
|
|
case <-recoveryDeadline.C:
|
|
recoveryTicker.Stop()
|
|
goto recoveryFinished
|
|
}
|
|
}
|
|
}
|
|
|
|
recoveryFinished:
|
|
if serverMetrics != nil {
|
|
if sample, scrapeErr := serverMetrics.scrape(ctx); scrapeErr == nil {
|
|
finalServerMetrics = sample
|
|
} else {
|
|
finalServerMetrics = nil
|
|
}
|
|
}
|
|
steadyRatio := float64(0)
|
|
if steadySamples > 0 {
|
|
steadyRatio = float64(steadyReadySum) / float64(steadySamples*len(workers))
|
|
}
|
|
report := &RunReport{
|
|
Version: 2, StartedAt: startedAt, LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(),
|
|
RequestedDuration: cfg.Duration.String(), RecoveryDuration: cfg.RecoveryDuration.String(),
|
|
ExpectedSessions: len(workers), PeakReadySessions: peakReady, FinalReadySessions: countWorkerState(workers, workerReady),
|
|
ConnectionAttempts: counters.connectionAttempts.Load(), Reconnects: counters.reconnects.Load(),
|
|
Disconnects: counters.disconnects.Load(), UpdatesReceived: counters.updates.Load(), DownloadedBytes: counters.downloadBytes.Load(),
|
|
WorkerFatalErrors: counters.fatalErrors.Load(), Operations: metrics.report(),
|
|
BaselineServerMetrics: baselineServerMetrics, FinalServerMetrics: finalServerMetrics,
|
|
ServerMetricsScrapes: serverMetrics.successes(), ServerMetricsErrors: serverMetrics.failures(),
|
|
SteadySamples: steadySamples, SteadyReadyRatio: steadyRatio, MinSteadyReadySessions: steadyReadyMinimum,
|
|
}
|
|
evaluateReport(report, cfg)
|
|
if err := WriteReport(cfg.ReportPath, report); err != nil {
|
|
return nil, err
|
|
}
|
|
return report, nil
|
|
}
|
|
|
|
func primaryTargets(records []SessionRecord) []SessionRecord {
|
|
byAccount := make(map[int]SessionRecord)
|
|
for _, record := range records {
|
|
if existing, ok := byAccount[record.AccountIndex]; !ok || record.DeviceIndex < existing.DeviceIndex {
|
|
byAccount[record.AccountIndex] = record
|
|
}
|
|
}
|
|
maxAccount := -1
|
|
for account := range byAccount {
|
|
maxAccount = max(maxAccount, account)
|
|
}
|
|
targets := make([]SessionRecord, maxAccount+1)
|
|
for account, record := range byAccount {
|
|
targets[account] = record
|
|
}
|
|
return targets
|
|
}
|
|
|
|
func minimumOpenFiles(sessions int) int {
|
|
if sessions < 0 {
|
|
sessions = 0
|
|
}
|
|
// gotd may transiently hold primary, PFS and replacement sockets together;
|
|
// retain fixed room for the process, resolver, event/report files and scrapes.
|
|
return 256 + sessions*6
|
|
}
|
|
|
|
func runOfflineWindow(ctx context.Context, workers []*loadWorker, fraction float64, at, duration time.Duration, events *eventWriter) {
|
|
timer := time.NewTimer(at)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timer.C:
|
|
}
|
|
count := int(math.Ceil(float64(len(workers)) * fraction))
|
|
selected := make([]*loadWorker, 0, count)
|
|
for i, worker := range workers {
|
|
if i%len(workers) < count {
|
|
worker.setOnline(false)
|
|
selected = append(selected, worker)
|
|
}
|
|
}
|
|
events.write(map[string]any{"type": "offline_start", "at": time.Now().UTC(), "sessions": len(selected)})
|
|
timer.Reset(duration)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timer.C:
|
|
}
|
|
for _, worker := range selected {
|
|
worker.setOnline(true)
|
|
}
|
|
events.write(map[string]any{"type": "offline_end", "at": time.Now().UTC(), "sessions": len(selected)})
|
|
}
|
|
|
|
func countWorkerState(workers []*loadWorker, state int32) int {
|
|
count := 0
|
|
for _, worker := range workers {
|
|
if worker.state.Load() == state {
|
|
count++
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
|
|
func writeSample(ctx context.Context, events *eventWriter, phase string, workers []*loadWorker, metrics *metricSet, counters *harnessCounters, server *serverMetricsClient) map[string]float64 {
|
|
var mem runtime.MemStats
|
|
runtime.ReadMemStats(&mem)
|
|
serverValues, scrapeErr := server.scrape(ctx)
|
|
value := map[string]any{
|
|
"type": "sample", "phase": phase, "at": time.Now().UTC(),
|
|
"workers": map[string]int{
|
|
"connecting": countWorkerState(workers, workerConnecting), "ready": countWorkerState(workers, workerReady),
|
|
"disconnected": countWorkerState(workers, workerDisconnected), "offline": countWorkerState(workers, workerOffline),
|
|
"stopped": countWorkerState(workers, workerStopped),
|
|
},
|
|
"client_runtime": map[string]uint64{"goroutines": uint64(runtime.NumGoroutine()), "heap_alloc_bytes": mem.HeapAlloc, "sys_bytes": mem.Sys},
|
|
"connections": map[string]uint64{
|
|
"attempts": counters.connectionAttempts.Load(), "reconnects": counters.reconnects.Load(),
|
|
"disconnects": counters.disconnects.Load(), "fatal_errors": counters.fatalErrors.Load(),
|
|
"downloaded_bytes": counters.downloadBytes.Load(),
|
|
},
|
|
"operations": metrics.report(), "server_metrics": serverValues,
|
|
}
|
|
if scrapeErr != nil {
|
|
value["server_metrics_error"] = classifyError(scrapeErr)
|
|
}
|
|
events.write(value)
|
|
return serverValues
|
|
}
|
|
|
|
func evaluateReport(report *RunReport, cfg RunConfig) {
|
|
requiredReady := int(math.Ceil(float64(report.ExpectedSessions) * cfg.MinimumReadyRatio))
|
|
if report.PeakReadySessions < requiredReady {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("peak ready sessions %d below required %d", report.PeakReadySessions, requiredReady))
|
|
}
|
|
if report.SteadySamples == 0 {
|
|
report.Failures = append(report.Failures, "no post-ramp steady-state samples were collected")
|
|
} else if report.SteadyReadyRatio < cfg.MinimumReadyRatio {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("steady ready ratio %.4f below required %.4f", report.SteadyReadyRatio, cfg.MinimumReadyRatio))
|
|
}
|
|
if report.WorkerFatalErrors > 0 {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("worker fatal errors: %d", report.WorkerFatalErrors))
|
|
}
|
|
for name, operation := range report.Operations {
|
|
if operation.FloodWaits > 0 {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("%s returned FLOOD_WAIT %d times", name, operation.FloodWaits))
|
|
}
|
|
unexpectedErrors := operation.Errors
|
|
if cfg.ExpectServerRestart {
|
|
unexpectedErrors -= min(unexpectedErrors, operation.ConnectionErrors)
|
|
}
|
|
if unexpectedErrors > 0 {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("%s returned %d unexpected non-cancel errors", name, unexpectedErrors))
|
|
}
|
|
}
|
|
if cfg.ExpectServerRestart && report.Reconnects < uint64(requiredReady) {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("server restart expected at least %d reconnect attempts, observed %d", requiredReady, report.Reconnects))
|
|
}
|
|
if report.FinalServerMetrics != nil && report.BaselineServerMetrics != nil && cfg.RecoveryDuration > 0 {
|
|
checks := []string{
|
|
"telesrv_mtproto_raw_connections", "telesrv_mtproto_logical_sessions",
|
|
"telesrv_mtproto_logical_outbox_bytes", "telesrv_mtproto_pending_push_bytes",
|
|
"telesrv_mtproto_outbound_tracked_bytes", "telesrv_mtproto_rpc_execution_owners",
|
|
"telesrv_mtproto_rpc_execution_reserved_entries", "telesrv_mtproto_rpc_execution_receipts",
|
|
"telesrv_mtproto_rpc_execution_receipt_budget_bytes", "telesrv_mtproto_rpc_execution_subscribers",
|
|
}
|
|
for _, name := range checks {
|
|
baseline := metricValue(report.BaselineServerMetrics, name)
|
|
final := metricValue(report.FinalServerMetrics, name)
|
|
if final > baseline {
|
|
report.Failures = append(report.Failures, fmt.Sprintf("server retained %.0f above baseline %.0f in %s after recovery", final-baseline, baseline, name))
|
|
}
|
|
}
|
|
}
|
|
if strings.TrimSpace(cfg.ServerMetricsURL) != "" && report.ServerMetricsScrapes == 0 {
|
|
report.Failures = append(report.Failures, "server metrics endpoint produced no successful scrapes")
|
|
}
|
|
if strings.TrimSpace(cfg.ServerMetricsURL) != "" && report.FinalServerMetrics == nil {
|
|
report.Failures = append(report.Failures, "final post-recovery server metrics scrape failed")
|
|
}
|
|
if strings.TrimSpace(cfg.ServerMetricsURL) != "" && report.BaselineServerMetrics == nil {
|
|
report.Failures = append(report.Failures, "pre-load server metrics baseline scrape failed")
|
|
}
|
|
report.Pass = len(report.Failures) == 0
|
|
}
|
|
|
|
func metricValue(values map[string]float64, name string) float64 {
|
|
var total float64
|
|
for key, value := range values {
|
|
if key == name || strings.HasPrefix(key, name+"{") {
|
|
total += value
|
|
}
|
|
}
|
|
return total
|
|
}
|
|
|
|
func classifyError(err error) string {
|
|
if err == nil {
|
|
return "ok"
|
|
}
|
|
if errors.Is(err, context.Canceled) {
|
|
return "canceled"
|
|
}
|
|
if errors.Is(err, context.DeadlineExceeded) {
|
|
return "timeout"
|
|
}
|
|
if errors.Is(err, mtproto.ErrPFSReconnectRequired) || errors.Is(err, mtproto.ErrPFSDropKeysRequired) || errors.Is(err, mtproto.ErrTransportNotReady) {
|
|
return "connection"
|
|
}
|
|
if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || errors.Is(err, pool.ErrConnDead) || errors.Is(err, tdrpc.ErrEngineClosed) ||
|
|
errors.Is(err, syscall.ECONNABORTED) || errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ECONNRESET) ||
|
|
errors.Is(err, syscall.EHOSTUNREACH) || errors.Is(err, syscall.ENETUNREACH) || errors.Is(err, syscall.EPIPE) {
|
|
return "connection"
|
|
}
|
|
var netErr net.Error
|
|
if errors.As(err, &netErr) {
|
|
if netErr.Timeout() {
|
|
return "timeout"
|
|
}
|
|
return "connection"
|
|
}
|
|
message := strings.ToUpper(err.Error())
|
|
switch {
|
|
case strings.Contains(message, "FLOOD_WAIT"):
|
|
return "flood_wait"
|
|
case strings.Contains(message, "ENCRYPTED_MESSAGE_INVALID"):
|
|
return "encrypted_message_invalid"
|
|
case strings.Contains(message, "AUTH_KEY"):
|
|
return "auth_key"
|
|
case strings.Contains(message, "CONNECTION"), strings.Contains(message, "CONNECT"), strings.Contains(message, "EOF"),
|
|
strings.Contains(message, "ENGINE WAS CLOSED"), strings.Contains(message, "BROKEN PIPE"),
|
|
strings.Contains(message, "CLOSED NETWORK"), strings.Contains(message, "NO ROUTE TO HOST"),
|
|
strings.Contains(message, "NETWORK IS UNREACHABLE"):
|
|
return "connection"
|
|
default:
|
|
return "error"
|
|
}
|
|
}
|
|
|
|
// classifyErrorReason intentionally returns a finite vocabulary. It preserves
|
|
// enough transport/PFS evidence to diagnose a failed load without persisting
|
|
// raw error strings, addresses, auth-key IDs or request payloads.
|
|
func classifyErrorReason(err error) string {
|
|
if err == nil {
|
|
return "ok"
|
|
}
|
|
message := strings.ToUpper(err.Error())
|
|
switch {
|
|
case errors.Is(err, mtproto.ErrPFSDropKeysRequired):
|
|
return "pfs_drop_keys"
|
|
case errors.Is(err, mtproto.ErrPFSReconnectRequired):
|
|
return "pfs_reconnect"
|
|
case errors.Is(err, mtproto.ErrTransportNotReady):
|
|
return "transport_not_ready"
|
|
case strings.Contains(message, "TOO MANY OPEN FILES"):
|
|
return "file_descriptor_limit"
|
|
case strings.Contains(message, "PFS RECONNECT"):
|
|
return "pfs_reconnect"
|
|
case strings.Contains(message, "AUTH KEY NOT FOUND"), strings.Contains(message, "AUTH_KEY_NOT_FOUND"), strings.Contains(message, "PROTOCOL ERROR 404"):
|
|
return "auth_key_not_found"
|
|
case strings.Contains(message, "ENCRYPTED_MESSAGE_INVALID"):
|
|
return "encrypted_message_invalid"
|
|
case strings.Contains(message, "FINGERPRINT"):
|
|
return "rsa_fingerprint"
|
|
case strings.Contains(message, "CONNECTION REFUSED"):
|
|
return "connection_refused"
|
|
case strings.Contains(message, "CONNECTION RESET"):
|
|
return "connection_reset"
|
|
case strings.Contains(message, "NETWORK IS UNREACHABLE"), strings.Contains(message, "NO ROUTE TO HOST"):
|
|
return "network_unreachable"
|
|
case strings.Contains(message, "NO SUCH HOST"):
|
|
return "dns"
|
|
case strings.Contains(message, "BROKEN PIPE"):
|
|
return "broken_pipe"
|
|
case strings.Contains(message, "EOF"):
|
|
return "eof"
|
|
case errors.Is(err, context.DeadlineExceeded):
|
|
return "timeout"
|
|
case errors.Is(err, context.Canceled):
|
|
return "canceled"
|
|
default:
|
|
return classifyError(err)
|
|
}
|
|
}
|