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/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) 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