From c3e1bdd5c32a3e38ef4f537b3e4e07cd07894b69 Mon Sep 17 00:00:00 2001 From: sssciel Date: Tue, 18 Aug 2026 13:35:19 +0300 Subject: [PATCH 1/2] status: add parallel status execution Previously, the status call was called linearly on all instances, which could lead to long delays. Now the calls are parallel. Also status eval requests could stuck, so separate flag --instance-timeout was added. Part of TNTP-6354 --- CHANGELOG.md | 6 +- cli/cmd/status.go | 13 +++- cli/connector/connector.go | 86 +++++++++++++++++------ cli/connector/connector_test.go | 81 ++++++++++++++++++++++ cli/status/status.go | 118 +++++++++++++++++++++++++------- cli/status/status_test.go | 117 +++++++++++++++++++++++++++++++ go.mod | 2 +- 7 files changed, 375 insertions(+), 48 deletions(-) create mode 100644 cli/connector/connector_test.go create mode 100644 cli/status/status_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 46231ac35..03c9779ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,10 +9,14 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added -### Changed +- `status`: added `--instance-timeout` flag to bound how long collecting a single + instance's status may take. ### Fixed +- `status`: replication errors were ignored. +- `status`: status requests could stuck into deadlock without timeout. + ## [2.14.0] - 2026-08-06 This release introduces cluster backup and restore: `tt backup` plans a diff --git a/cli/cmd/status.go b/cli/cmd/status.go index 8dc6e88ee..a485617d5 100644 --- a/cli/cmd/status.go +++ b/cli/cmd/status.go @@ -2,6 +2,7 @@ package cmd import ( "fmt" + "time" "github.com/spf13/cobra" "github.com/tarantool/tt/cli/cmd/internal" @@ -18,6 +19,10 @@ type statusOpts struct { details bool // Deprecated: use --format instead. pretty bool + // instanceTimeout bounds how long collecting a single instance's status may + // take, so that one stuck instance can't hang the whole command. Zero disables + // the timeout. + instanceTimeout time.Duration } var opts statusOpts @@ -61,6 +66,9 @@ Columns: statusCmd.Flags().BoolVarP(&opts.pretty, "pretty", "p", false, "output a pretty-formatted table (deprecated, use --format instead)") statusCmd.Flags().MarkDeprecated("pretty", "use --format instead") + statusCmd.Flags().DurationVar(&opts.instanceTimeout, "instance-timeout", + status.DefaultInstanceTimeout, + "timeout for collecting a single instance's status; 0 disables the timeout") return statusCmd } @@ -106,5 +114,8 @@ func internalStatusModule(cmdCtx *cmdcontext.CmdCtx, args []string) error { printer = status.NewTablePrinter(status.WithDetails(opts.details)) } - return status.Status(runningCtx, printer) + if err := status.Status(runningCtx, printer, opts.instanceTimeout); err != nil { + return fmt.Errorf("failed to get status: %w", err) + } + return nil } diff --git a/cli/connector/connector.go b/cli/connector/connector.go index 4e555adb0..81d7eed13 100644 --- a/cli/connector/connector.go +++ b/cli/connector/connector.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "runtime" + "sync" "time" "github.com/tarantool/go-tarantool" @@ -18,6 +19,59 @@ const ( maxSocketPathMac = 106 ) +// connectMutex serializes connections that depend on the process-wide working +// directory. prepareUnixAddress may temporarily change it to shorten a socket path. +var connectMutex sync.Mutex + +// unixSocketPathLimit returns the maximum socket path length for the current OS. +func unixSocketPathLimit() int { + if runtime.GOOS == "darwin" { + return maxSocketPathMac + } + return maxSocketPathLinux +} + +// prepareUnixAddress prepares a Unix socket address for use with Tarantool. +func prepareUnixAddress(address string) (string, func(), error) { + maxSocketPath := unixSocketPathLimit() + + pathNeedsShortening := len(address)+1 > maxSocketPath + if filepath.IsAbs(address) && !pathNeedsShortening { + return address, nil, nil + } + + shortAddress := "./" + filepath.Base(address) + if pathNeedsShortening && len(shortAddress)+1 > maxSocketPath { + return "", nil, fmt.Errorf("socket name is longer than %d symbols: %s", + maxSocketPath-3, filepath.Base(address)) + } + + // Relative paths also depend on the process-wide working directory. + connectMutex.Lock() // unlock in cleanup. + + if !pathNeedsShortening { + return address, connectMutex.Unlock, nil + } + + workDir, err := os.Getwd() + if err != nil { + connectMutex.Unlock() + return "", nil, fmt.Errorf("failed to get working directory: %w", err) + } + + if err := os.Chdir(filepath.Dir(address)); err != nil { + connectMutex.Unlock() + return "", nil, fmt.Errorf("failed to change directory to socket directory: %w", err) + } + + cleanup := func() { + _ = os.Chdir(workDir) + connectMutex.Unlock() + } + + return shortAddress, cleanup, nil +} + // RequestOpts describes the parameters of a request to be executed. type RequestOpts struct { // PushCallback is the cb that will be called when a "push" message is received. @@ -43,29 +97,20 @@ type Connector interface { // Connect connects to the tarantool instance according to options. func Connect(opts ConnectOpts) (Connector, error) { - // It became common that address is longer than 108 symbols(sun_path limit). - // To reduce length of address we use relative path - // with chdir into a directory of socket. - // e.g foo/bar/123.sock -> ./123.sock - workDir, err := os.Getwd() - if err != nil { - return nil, err - } - - maxSocketPath := maxSocketPathLinux - if runtime.GOOS == "darwin" { - maxSocketPath = maxSocketPathMac - } + if opts.Network == "unix" { + address, cleanup, err := prepareUnixAddress(opts.Address) + if err != nil { + return nil, fmt.Errorf("failed to prepare unix socket address: %w", err) + } - if _, err := os.Stat(opts.Address); err == nil { - os.Chdir(filepath.Dir(opts.Address)) - opts.Address = "./" + filepath.Base(opts.Address) - if len(opts.Address)+1 > maxSocketPath { - return nil, fmt.Errorf("socket name is longer than %d symbols: %s", - maxSocketPath-3, filepath.Base(opts.Address)) + if cleanup != nil { + defer cleanup() } - defer os.Chdir(workDir) + + // Use the short address if it was prepared. + opts.Address = address } + // Connect to specified address. greetingConn, err := net.Dial(opts.Network, opts.Address) if err != nil { @@ -85,6 +130,7 @@ func Connect(opts ConnectOpts) (Connector, error) { protocol = BinaryProtocol transport = "ssl" } else { + greetingConn.Close() return nil, fmt.Errorf("failed to get protocol: %s", err) } } else if ssl { diff --git a/cli/connector/connector_test.go b/cli/connector/connector_test.go new file mode 100644 index 000000000..62dae5d3d --- /dev/null +++ b/cli/connector/connector_test.go @@ -0,0 +1,81 @@ +package connector + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestPrepareUnixAddressShortAbsolutePath(t *testing.T) { + address := filepath.Join(t.TempDir(), "instance.sock") + if len(address)+1 > unixSocketPathLimit() { + t.Skip("temporary directory path is too long for this test") + } + + prepared, cleanup, err := prepareUnixAddress(address) + + require.NoError(t, err) + require.Equal(t, address, prepared) + require.Nil(t, cleanup) +} + +func TestPrepareUnixAddressRelativePath(t *testing.T) { + const address = "run/instance.sock" + + prepared, cleanup, err := prepareUnixAddress(address) + require.NoError(t, err) + require.NotNil(t, cleanup) + defer cleanup() + + require.Equal(t, address, prepared) +} + +func TestPrepareUnixAddressLongPath(t *testing.T) { + const socketName = "instance.sock" + + originalWorkDir, err := os.Getwd() + require.NoError(t, err) + + socketDir := t.TempDir() + for len(filepath.Join(socketDir, socketName))+1 <= unixSocketPathLimit() { + socketDir = filepath.Join(socketDir, strings.Repeat("d", 32)) + } + require.NoError(t, os.MkdirAll(socketDir, 0o755)) + + prepared, cleanup, err := prepareUnixAddress(filepath.Join(socketDir, socketName)) + require.NoError(t, err) + require.NotNil(t, cleanup) + + cleanedUp := false + defer func() { + if !cleanedUp { + cleanup() + } + }() + + currentWorkDir, err := os.Getwd() + require.NoError(t, err) + require.Equal(t, filepath.Clean(socketDir), currentWorkDir) + require.Equal(t, "./"+socketName, prepared) + + cleanup() + cleanedUp = true + + currentWorkDir, err = os.Getwd() + require.NoError(t, err) + require.Equal(t, originalWorkDir, currentWorkDir) +} + +func TestPrepareUnixAddressLongSocketName(t *testing.T) { + socketName := strings.Repeat("s", unixSocketPathLimit()) + address := filepath.Join(t.TempDir(), socketName) + + prepared, cleanup, err := prepareUnixAddress(address) + + require.ErrorContains(t, err, "socket name is longer") + require.Empty(t, prepared) + require.Nil(t, cleanup) +} diff --git a/cli/status/status.go b/cli/status/status.go index d00a6d18e..df229bdb0 100644 --- a/cli/status/status.go +++ b/cli/status/status.go @@ -1,14 +1,18 @@ package status import ( + "context" _ "embed" "fmt" "strings" + "sync" + "time" "github.com/mitchellh/mapstructure" "github.com/tarantool/tt/cli/connector" "github.com/tarantool/tt/cli/process_utils" "github.com/tarantool/tt/cli/running" + "golang.org/x/sync/semaphore" ) // InstanceStatusPrinter interface defines methods to output instance status information. @@ -16,6 +20,13 @@ type InstanceStatusPrinter interface { Print(instances map[string]*instanceStatus) error } +// maxParallelStatusRequests limits the number of concurrent instance status checks. +const maxParallelStatusRequests = 128 + +// DefaultInstanceTimeout is the default timeout for collecting a single instance's +// status, used when the caller does not override it (e.g. via --instance-timeout). +const DefaultInstanceTimeout = 5 * time.Second + //go:embed lua/instance_state.lua var instanceInfoLuaScript string @@ -138,6 +149,7 @@ func processConfigInfo(instStatus *instanceStatus, instanceState rawInstanceStat if len(instanceState.ConfigInfo.Alerts) == 0 { return } + for _, alert := range instanceState.ConfigInfo.Alerts { severity := severityWarning if alert.Type == "error" { @@ -149,7 +161,7 @@ func processConfigInfo(instStatus *instanceStatus, instanceState rawInstanceStat // collectInstanceState connects to an instance and collects its state. func collectInstanceState(run running.InstanceCtx, fullInstanceName string, - instStatus *instanceStatus, + instStatus *instanceStatus, instanceTimeout time.Duration, ) (rawInstanceState, error) { var instanceState rawInstanceState @@ -167,8 +179,10 @@ func collectInstanceState(run running.InstanceCtx, fullInstanceName string, fullInstanceName, err) } + defer conn.Close() + res, err := conn.Eval(filterComments(instanceInfoLuaScript), []any{}, - connector.RequestOpts{}) + connector.RequestOpts{ReadTimeout: instanceTimeout}) if err != nil { instStatus.addAlert(fmt.Sprintf( "Error while executing Lua script on instance %s: %v", @@ -195,34 +209,88 @@ func collectInstanceState(run running.InstanceCtx, fullInstanceName string, return instanceState, nil } -// Status writes the status as a table. -func Status(runningCtx running.RunningCtx, printer InstanceStatusPrinter) error { - instances := make(instanceStatusMap) - uuid2name := map[string]string{} - for _, run := range runningCtx.Instances { - fullInstanceName := running.GetAppInstanceName(run) - instStatus := newInstanceStatus() - instStatus.procStatus = running.Status(&run) - instStatus.Status = instStatus.procStatus.Status - instances[fullInstanceName] = &instStatus +// statusResult holds the result of a status check for an instance. +type statusResult struct { + name string + // Since Tarantool 2.x doesn't support instance names, only UUIDs are available. + // To make the alerts more readable, we map the UUIDs to instance names. + uuid string + status *instanceStatus +} - if instStatus.procStatus.Code == process_utils.ProcessRunningCode { - instStatus.PID = &instStatus.procStatus.PID +func collectStatuses(instances []running.InstanceCtx, + instanceTimeout time.Duration, +) <-chan statusResult { + statuses := make(chan statusResult, len(instances)) + sem := semaphore.NewWeighted(maxParallelStatusRequests) + ctx := context.Background() + + var wg sync.WaitGroup + for _, instance := range instances { + if err := sem.Acquire(ctx, 1); err != nil { + break } - instanceState, err := collectInstanceState(run, fullInstanceName, &instStatus) - if err != nil { - continue - } + wg.Go(func() { + defer sem.Release(1) + statuses <- processStatusForInstance(instance, instanceTimeout) + }) + } + + go func() { + wg.Wait() + close(statuses) + }() + + return statuses +} - // Since Tarantool 2.x doesn't support instance names, only UUIDs are available. - // To make the alerts more readable, we map the UUIDs to instance names. - uuid2name[instanceState.UUID] = fullInstanceName +func applyInstanceState(instStatus *instanceStatus, instanceState rawInstanceState) { + processConfigInfo(instStatus, instanceState) + instStatus.Mode = instanceState.ReadOnly + instStatus.Config = instanceState.ConfigInfo.Status + instStatus.Box = instanceState.BoxStatus + instStatus.rawReplicationInfo = instanceState.ReplicationInfo +} + +func processStatusForInstance(instance running.InstanceCtx, + instanceTimeout time.Duration, +) statusResult { + instStatus := newInstanceStatus() + instStatus.procStatus = running.Status(&instance) + instStatus.Status = instStatus.procStatus.Status + if instStatus.procStatus.Code == process_utils.ProcessRunningCode { + instStatus.PID = &instStatus.procStatus.PID + } + + fullInstanceName := running.GetAppInstanceName(instance) + + instanceState, err := collectInstanceState(instance, fullInstanceName, &instStatus, + instanceTimeout) + if err != nil { + return statusResult{name: fullInstanceName, status: &instStatus} + } - processConfigInfo(&instStatus, instanceState) - instStatus.Mode = instanceState.ReadOnly - instStatus.Config = instanceState.ConfigInfo.Status - instStatus.Box = instanceState.BoxStatus + applyInstanceState(&instStatus, instanceState) + + return statusResult{name: fullInstanceName, uuid: instanceState.UUID, status: &instStatus} +} + +// Status writes the status as a table. instanceTimeout bounds how long collecting +// a single instance's status may take (e.g. an instance stuck processing requests); +// zero means no timeout. +func Status(runningCtx running.RunningCtx, printer InstanceStatusPrinter, + instanceTimeout time.Duration, +) error { + instances := make(instanceStatusMap) + uuid2name := map[string]string{} + statuses := collectStatuses(runningCtx.Instances, instanceTimeout) + + for instStatus := range statuses { + if instStatus.uuid != "" { + uuid2name[instStatus.uuid] = instStatus.name + } + instances[instStatus.name] = instStatus.status } for _, instStatus := range instances { diff --git a/cli/status/status_test.go b/cli/status/status_test.go new file mode 100644 index 000000000..897a14765 --- /dev/null +++ b/cli/status/status_test.go @@ -0,0 +1,117 @@ +package status + +import ( + "net" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/tarantool/tt/cli/running" +) + +func TestApplyInstanceStatePreservesReplicationInfo(t *testing.T) { + const ( + upstreamUUID = "11111111-1111-1111-1111-111111111111" + upstreamName = "app:master" + ) + + state := rawInstanceState{ + ReplicationInfo: []rawReplicationInfo{{ + UUID: upstreamUUID, + Upstream: upstream{ + Status: "disconnected", + Message: "connection refused", + }, + }}, + ConfigInfo: configInfo{Status: "ready"}, + ReadOnly: "RO", + BoxStatus: "running", + } + instStatus := newInstanceStatus() + + applyInstanceState(&instStatus, state) + processReplicationInfo(&instStatus, map[string]string{upstreamUUID: upstreamName}) + + require.Equal(t, "RO", instStatus.Mode) + require.Equal(t, "ready", instStatus.Config) + require.Equal(t, "running", instStatus.Box) + require.Equal(t, "disconnected", instStatus.Upstream) + require.Equal(t, state.ReplicationInfo, instStatus.rawReplicationInfo) + require.Equal(t, []instanceAlert{{ + Message: "[upstream][warning]: replication from instance with name \"app:master\" " + + "is in \"disconnected\" status: \"connection refused\"", + Severity: severityWarning, + }}, instStatus.Alerts) +} + +// startHangingConsoleServer starts a fake console (plain text protocol) server that +// sends a valid Tarantool greeting and then never responds to any request, simulating +// an instance whose main fiber is deadlocked. +func startHangingConsoleServer(t *testing.T) string { + t.Helper() + + socketPath := filepath.Join(t.TempDir(), "instance.sock") + ln, err := net.Listen("unix", socketPath) + require.NoError(t, err) + + done := make(chan struct{}) + t.Cleanup(func() { + close(done) + ln.Close() + }) + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + + greeting := make([]byte, 128) + copy(greeting, []byte("Tarantool 2.11.0 (Lua console)\n")) + if _, err := conn.Write(greeting); err != nil { + return + } + + // Never read or write again: the instance is deadlocked and can't + // process the eval request at all. Block until the test cleans up. + <-done + }() + + return socketPath +} + +func TestCollectInstanceStateRespectsInstanceTimeout(t *testing.T) { + socketPath := startHangingConsoleServer(t) + + run := running.InstanceCtx{ConsoleSocket: socketPath} + instStatus := newInstanceStatus() + + const instanceTimeout = 100 * time.Millisecond + start := time.Now() + _, err := collectInstanceState(run, "test-instance", &instStatus, instanceTimeout) + elapsed := time.Since(start) + + require.Error(t, err) + require.ErrorContains(t, err, "test-instance") + require.ErrorContains(t, err, "i/o timeout") + require.Less(t, elapsed, instanceTimeout+2*time.Second, + "collectInstanceState should be bounded by instanceTimeout, not hang forever") +} + +func TestProcessStatusForInstanceRespectsInstanceTimeout(t *testing.T) { + socketPath := startHangingConsoleServer(t) + + run := running.InstanceCtx{ConsoleSocket: socketPath} + + const instanceTimeout = 100 * time.Millisecond + start := time.Now() + result := processStatusForInstance(run, instanceTimeout) + elapsed := time.Since(start) + + require.NotNil(t, result.status) + require.NotEmpty(t, result.status.Alerts) + require.Less(t, elapsed, instanceTimeout+2*time.Second, + "processStatusForInstance should be bounded by instanceTimeout, not hang forever") +} diff --git a/go.mod b/go.mod index 72d600f8d..1d2064576 100644 --- a/go.mod +++ b/go.mod @@ -48,6 +48,7 @@ require ( go.etcd.io/etcd/tests/v3 v3.6.8 golang.org/x/crypto v0.51.0 golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa + golang.org/x/sync v0.20.0 golang.org/x/sys v0.45.0 golang.org/x/term v0.43.0 google.golang.org/grpc v1.79.3 @@ -179,7 +180,6 @@ require ( go.uber.org/zap v1.27.1 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/net v0.53.0 // indirect - golang.org/x/sync v0.20.0 // indirect golang.org/x/text v0.37.0 // indirect golang.org/x/time v0.14.0 // indirect golang.org/x/tools/godoc v0.1.0-deprecated // indirect From d8c10386faa8f213d6fbc248d60d87cf914b4f89 Mon Sep 17 00:00:00 2001 From: sssciel Date: Tue, 18 Aug 2026 14:50:01 +0300 Subject: [PATCH 2/2] running: optimize config reading time Previously, the cluster config was re-read from disk for each instance, which could lead to huge delays on hundreds of instances. Now the cluster config is read once. Closes TNTP-6354 --- cli/running/running.go | 43 ++++++++++++++++++++++++++++++------------ 1 file changed, 31 insertions(+), 12 deletions(-) diff --git a/cli/running/running.go b/cli/running/running.go index 3361c6584..9371f308d 100644 --- a/cli/running/running.go +++ b/cli/running/running.go @@ -335,13 +335,13 @@ func findInstanceScriptInAppDir(appDir, instName, clusterCfgPath, defaultScript return script, nil } -// loadInstanceConfig load instance configuration from cluster config. -func loadInstanceConfig(configPath, instName string, +// loadClusterConfig reads and parses a cluster config. +func loadClusterConfig(configPath string, integrityCtx integrity.IntegrityCtx, -) (libcluster.InstanceConfig, error) { - var instCfg libcluster.InstanceConfig +) (libcluster.ClusterConfig, error) { + var clusterCfg libcluster.ClusterConfig if configPath == "" { - return instCfg, nil + return clusterCfg, nil } var dataCollectors libcluster.DataCollectorFactory @@ -349,7 +349,7 @@ func loadInstanceConfig(configPath, instName string, if err == integrity.ErrNotConfigured { dataCollectors = libcluster.NewDataCollectorFactory() } else if err != nil { - return instCfg, + return clusterCfg, fmt.Errorf("failed to create collectors with integrity check: %w", err) } else { dataCollectors = libcluster.NewIntegrityDataCollectorFactory(checkFunc, @@ -359,12 +359,24 @@ func loadInstanceConfig(configPath, instName string, } collectors := libcluster.NewCollectorFactory(dataCollectors) - clusterCfg, err := cluster.GetClusterConfig(collectors, configPath) + clusterCfg, err = cluster.GetClusterConfig(collectors, configPath) if err != nil { - return instCfg, err + return clusterCfg, fmt.Errorf("failed to get cluster config: %w", err) + } + return clusterCfg, nil +} + +// loadInstanceConfig derives an instance configuration from an +// cluster config. +func loadInstanceConfig(clusterCfg libcluster.ClusterConfig, configPath, + instName string, +) (libcluster.InstanceConfig, error) { + if configPath == "" { + return libcluster.InstanceConfig{}, nil } - if instCfg, err = cluster.GetInstanceConfig(clusterCfg, instName); err != nil { - return instCfg, err + instCfg, err := cluster.GetInstanceConfig(clusterCfg, instName) + if err != nil { + return instCfg, fmt.Errorf("failed to get instance config: %w", err) } return instCfg, nil } @@ -418,6 +430,13 @@ func collectInstancesFromAppDir(appDir, selectedInstName string, if err != nil { return nil, err } + + clusterCfg, err := loadClusterConfig(appDirFiles.clusterCfgPath, integrityCtx) + if err != nil && (loadConfig == ConfigLoadAll || loadConfig == ConfigLoadCluster) { + return nil, fmt.Errorf("error loading cluster configuration from config %q: %w", + appDirFiles.clusterCfgPath, err) + } + log.Debug("Processing application instances file") instances := []InstanceCtx{} for inst := range instParams { @@ -433,8 +452,8 @@ func collectInstancesFromAppDir(appDir, selectedInstName string, } log.Debugf("Instance %q", instance.InstName) - instance.Configuration, err = loadInstanceConfig(instance.ClusterConfigPath, - instance.InstName, integrityCtx) + instance.Configuration, err = loadInstanceConfig(clusterCfg, + instance.ClusterConfigPath, instance.InstName) if err != nil && (loadConfig == ConfigLoadAll || loadConfig == ConfigLoadCluster) { return instances, fmt.Errorf("error loading instance %q configuration from "+ "config %q: %w", instance.InstName, instance.ClusterConfigPath, err)