chore: remove unused code (#159)

moflotas · eguguchkin · web-flow · commit 54db16aecd36 · 2025-10-21T19:27:19.000+03:00
Co-authored-by: Che &lt;107068454+eguguchkin@users.noreply.github.com&gt;
diff --git a/cmd/stress-search/main.go b/cmd/stress-search/main.go
@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/alecthomas/units"
 
@@ -59,10 +60,9 @@ func main() {
 		cancel()
 	}()
 
-	conn, err := grpc.DialContext(
-		ctx,
+	conn, err := grpc.NewClient(
 		*endpoint,
-		grpc.WithInsecure(),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithStatsHandler(&tracing.ClientHandler{}),
 	)
 
diff --git a/frac/active_index.go b/frac/active_index.go
@@ -182,7 +182,7 @@ func (si *activeTokenIndex) GetValByTID(tid uint32) []byte {
 }
 
 func (si *activeTokenIndex) GetTIDsByTokenExpr(t parser.Token) ([]uint32, error) {
-	return si.tokenList.FindPattern(si.ctx, t, nil)
+	return si.tokenList.FindPattern(si.ctx, t)
 }
 
 func (si *activeTokenIndex) GetLIDsFromTIDs(tids []uint32, _ lids.Counter, minLID, maxLID uint32, order seq.DocsOrder) []node.Node {
diff --git a/frac/active_token_list.go b/frac/active_token_list.go
@@ -184,7 +184,7 @@ func (tl *TokenList) getTokenProvider(field string) *activeTokenProvider {
 	}
 }
 
-func (tl *TokenList) FindPattern(ctx context.Context, t parser.Token, tids []uint32) ([]uint32, error) {
+func (tl *TokenList) FindPattern(ctx context.Context, t parser.Token) ([]uint32, error) {
 	field := parser.GetField(t)
 	tp := tl.getTokenProvider(field)
 	tids, err := pattern.Search(ctx, t, tp)
diff --git a/logger/util.go b/logger/util.go
diff --git a/prepare/prepare.go b/prepare/prepare.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"io/ioutil"
 	_ "net/http/pprof"
 	"os"
 	"path/filepath"
@@ -94,7 +93,7 @@ func main() {
 		logger.Fatal("can't create dir for batches", zap.Error(err))
 	}
 	for i, batch := range batches {
-		err := ioutil.WriteFile(fmt.Sprintf("%s/%d.json", batchesDir, i), batch.Bytes(), 0o777)
+		err := os.WriteFile(fmt.Sprintf("%s/%d.json", batchesDir, i), batch.Bytes(), 0o777)
 		if err != nil {
 			logger.Fatal("can't write batch",
 				zap.Int("i", i),
diff --git a/proxyapi/grpc_fetch_test.go b/proxyapi/grpc_fetch_test.go
@@ -218,7 +218,6 @@ func TestGrpcV1_FetchLive(t *testing.T) {
 	}
 
 	for _, tt := range tests {
-		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
 			r := require.New(t)
diff --git a/proxyapi/grpc_main_test.go b/proxyapi/grpc_main_test.go
@@ -103,7 +103,6 @@ func TestMain(m *testing.M) {
 }
 
 func runGRPCServerWithClient(apiServer seqproxyapi.SeqProxyApiServer) (seqproxyapi.SeqProxyApiClient, func()) {
-	ctx := context.Background()
 	lis := bufconn.Listen(10 * 1024 * 1024)
 	server := grpc.NewServer(
 		grpc.StatsHandler(&tracing.ServerHandler{}),
@@ -121,9 +120,8 @@ func runGRPCServerWithClient(apiServer seqproxyapi.SeqProxyApiServer) (seqproxya
 		}
 		server.Stop()
 	}
-	conn, err := grpc.DialContext(
-		ctx,
-		"",
+	conn, err := grpc.NewClient(
+		"localhost",
 		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
 			return lis.Dial()
 		}),
diff --git a/proxyapi/http_bulk.go b/proxyapi/http_bulk.go
@@ -256,7 +256,7 @@ func (r *esBulkDocReader) readDoc() ([]byte, bool, error) {
 
 	// If this is a line that exceeds maxDocumentSize, skip it.
 	for isPrefix {
-		doc, isPrefix, err = r.r.ReadLine()
+		_, isPrefix, err = r.r.ReadLine()
 		if err != nil {
 			return nil, true, err
 		}
diff --git a/proxyapi/ingestor.go b/proxyapi/ingestor.go
@@ -47,38 +47,37 @@ type Ingestor struct {
 	isStopped atomic.Bool
 }
 
-func clientsFromConfig(ctx context.Context, config search.Config) (map[string]storeapi.StoreApiClient, error) {
+func clientsFromConfig(config search.Config) (map[string]storeapi.StoreApiClient, error) {
 	clients := map[string]storeapi.StoreApiClient{}
-	if err := appendClients(ctx, clients, config.HotStores.Shards); err != nil {
+	if err := appendClients(clients, config.HotStores.Shards); err != nil {
 		return nil, err
 	}
 	if config.HotReadStores != nil {
-		if err := appendClients(ctx, clients, config.HotReadStores.Shards); err != nil {
+		if err := appendClients(clients, config.HotReadStores.Shards); err != nil {
 			return nil, err
 		}
 	}
 	if config.WriteStores != nil {
-		if err := appendClients(ctx, clients, config.WriteStores.Shards); err != nil {
+		if err := appendClients(clients, config.WriteStores.Shards); err != nil {
 			return nil, err
 		}
 	}
 	if config.ReadStores != nil {
-		if err := appendClients(ctx, clients, config.ReadStores.Shards); err != nil {
+		if err := appendClients(clients, config.ReadStores.Shards); err != nil {
 			return nil, err
 		}
 	}
 	return clients, nil
 }
 
-func appendClients(ctx context.Context, clients map[string]storeapi.StoreApiClient, shards [][]string) error {
+func appendClients(clients map[string]storeapi.StoreApiClient, shards [][]string) error {
 	for _, shard := range shards {
 		for _, replica := range shard {
 			if _, has := clients[replica]; has {
 				continue
 			}
 			// this doesn't block, and if store is down, it will try to reconnect in background
-			conn, err := grpc.DialContext(
-				ctx,
+			conn, err := grpc.NewClient(
 				replica,
 				grpc.WithTransportCredentials(insecure.NewCredentials()),
 				grpc.WithStatsHandler(&tracing.ClientHandler{}),
@@ -132,7 +131,7 @@ func NewIngestor(config IngestorConfig, store *storeapiclient.Store) (*Ingestor,
 		config.Bulk.HotStores = stores.NewStoresFromString("memory", 1)
 		config.Search.HotStores = stores.NewStoresFromString("memory", 1)
 	} else {
-		clients, err = clientsFromConfig(ctx, config.Search)
+		clients, err = clientsFromConfig(config.Search)
 		if err != nil {
 			cancel()
 			return nil, fmt.Errorf("initialize clients: %s", err)
diff --git a/tests/common/byte_string.go b/tests/common/byte_string.go
@@ -1,16 +1,5 @@
 package common
 
-func ToBytesSlice(s []string) [][]byte {
-	if s == nil {
-		return nil
-	}
-	res := make([][]byte, 0, len(s))
-	for _, v := range s {
-		res = append(res, []byte(v))
-	}
-	return res
-}
-
 func ToStringSlice(s [][]byte) []string {
 	if s == nil {
 		return nil
@@ -21,25 +10,3 @@ func ToStringSlice(s [][]byte) []string {
 	}
 	return res
 }
-
-func ToBytesSlice2d(s [][]string) [][][]byte {
-	if s == nil {
-		return nil
-	}
-	res := make([][][]byte, 0, len(s))
-	for _, v := range s {
-		res = append(res, ToBytesSlice(v))
-	}
-	return res
-}
-
-func ToStringSlice2d(s [][][]byte) [][]string {
-	if s == nil {
-		return nil
-	}
-	res := make([][]string, 0, len(s))
-	for _, v := range s {
-		res = append(res, ToStringSlice(v))
-	}
-	return res
-}
diff --git a/tests/common/util.go b/tests/common/util.go
@@ -6,32 +6,11 @@ import (
 	"sync"
 	"testing"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"lukechampine.com/frand"
-
-	"github.com/ozontech/seq-db/seq"
 )
 
 var baseTmpDir string
 
-func MakeTouchTotal() *prometheus.CounterVec {
-	return prometheus.NewCounterVec(prometheus.CounterOpts{
-		Namespace: "seq_db_store",
-		Subsystem: "cache",
-		Name:      "touch_total",
-		Help:      "",
-	}, []string{"layer"})
-}
-
-func MakeHitsTotal() *prometheus.CounterVec {
-	return prometheus.NewCounterVec(prometheus.CounterOpts{
-		Namespace: "seq_db_store",
-		Subsystem: "cache",
-		Name:      "hits_total",
-		Help:      "",
-	}, []string{"layer"})
-}
-
 func CreateDir(path string) {
 	err := os.MkdirAll(path, 0o777)
 	if err != nil {
@@ -70,15 +49,6 @@ func GetTestTmpDir(t *testing.T) string {
 	return filepath.Join(CreateTempDir(), t.Name())
 }
 
-func IDs(ids ...int) []seq.ID {
-	r := make([]seq.ID, 0)
-	for _, n := range ids {
-		r = append(r, seq.SimpleID(n))
-	}
-
-	return r
-}
-
 func RandomString(minLen, maxLen int) string {
 	size := frand.Intn(maxLen-minLen+1) + minLen
 	res := make([]byte, size)
diff --git a/tests/integration_tests/sub_search_test.go b/tests/integration_tests/sub_search_test.go
@@ -32,12 +32,11 @@ func (s *IntegrationTestSuite) ingestData(env *setup.TestingEnv, from, to time.T
 
 		step := fractionTo.Sub(fractionFrom) / time.Duration(docsPerFraction)
 
-		for i, ts := 0, fractionFrom; i < docsPerFraction; i++ {
-			strTs := ts.Format(time.RFC3339)
-			origDocs = append(origDocs, fmt.Sprintf(`{"service":"x%d", "ts":%q}`, i, strTs))
-			ts, _ = time.Parse(time.RFC3339, strTs)
-			docsTimes = append(docsTimes, ts)
-			ts.Add(step)
+		for i, ts := 0, fractionFrom; i < docsPerFraction; i, ts = i+1, ts.Add(step) {
+			// using truncate, since RFC3339 is sub-second accurate
+			t := ts.Truncate(time.Second)
+			origDocs = append(origDocs, fmt.Sprintf(`{"service":"x%d", "ts":%q}`, i, t.Format(time.RFC3339)))
+			docsTimes = append(docsTimes, t)
 		}
 
 		setup.Bulk(s.T(), env.IngestorBulkAddr(), origDocs)
diff --git a/tests/setup/doc.go b/tests/setup/doc.go
@@ -90,16 +90,6 @@ func RandomWord() string {
 	return randomStringFromPool(words)
 }
 
-var fields = []string{
-	"service", "message", "trace_id", "source",
-	"zone", "level", "timestamp", "requestObject",
-	"", // will be random
-}
-
-func RandomField() string {
-	return randomStringFromPool(fields)
-}
-
 func RandomSymbol() byte {
 	symbols := "!@#$%^&*(){}[]:?/\\~;"
 	return symbols[frand.Intn(len(symbols))]
@@ -214,17 +204,6 @@ func DocsToStrings(docs []ExampleDoc) []string {
 	return docStr
 }
 
-func DocsFromStrings(docStr []string) []ExampleDoc {
-	docs := make([]ExampleDoc, len(docStr))
-	for i, doc := range docStr {
-		err := json.Unmarshal([]byte(doc), &docs[i])
-		if err != nil {
-			panic(err)
-		}
-	}
-	return docs
-}
-
 func splitRange(size int, callback func(from int, to int)) {
 	routines := max(1, min(size/10000, runtime.NumCPU())) // do parallel for large arrays
 
diff --git a/tests/setup/methods.go b/tests/setup/methods.go
@@ -26,19 +26,6 @@ func GenBuffer(docs []string) *bytes.Buffer {
 	return b
 }
 
-func GenBufferBytesReuse(docs [][]byte, b *bytes.Buffer) *bytes.Buffer {
-	for _, doc := range docs {
-		_, _ = b.WriteString(`{"index":"seq-db"}` + "\n")
-		_, _ = b.Write(doc)
-		_, _ = b.WriteString("\n")
-	}
-	return b
-}
-
-func GenBufferBytes(docs [][]byte) *bytes.Buffer {
-	return GenBufferBytesReuse(docs, bytes.NewBuffer(nil))
-}
-
 func BulkBuffer(t *testing.T, addr string, b *bytes.Buffer) {
 	r, err := http.Post(addr, "", b)
 	require.NoError(t, err, "should be no errors")
diff --git a/util/err.go b/util/err.go
@@ -1,9 +1,7 @@
 package util
 
 import (
-	"errors"
 	"fmt"
-	"runtime"
 	"runtime/debug"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -23,20 +21,6 @@ func (p *panicWrapper) Error() string {
 	return p.e.Error()
 }
 
-func IsRecoveredPanicError(e error) bool {
-	if e == nil {
-		return false
-	}
-
-	var p *panicWrapper
-	if errors.As(e, &p) {
-		return true
-	}
-
-	var re runtime.Error
-	return errors.As(e, &re)
-}
-
 func Recover(metric prometheus.Counter, err error) {
 	metric.Inc()
 	logger.Error(err.Error())
diff --git a/util/realloc_solver.go b/util/realloc_solver.go
@@ -24,10 +24,6 @@ type ReallocSolver struct {
 
 type ReallocSolverOpts func(*ReallocSolver)
 
-func ReallocSolverSize(statSize int) ReallocSolverOpts {
-	return func(r *ReallocSolver) { r.statSize = statSize }
-}
-
 func ReallocSolverLabel(label string) ReallocSolverOpts {
 	return func(r *ReallocSolver) { r.label = label }
 }

Original file line number	Diff line number	Diff line change
`@@ -182,7 +182,7 @@ func (si *activeTokenIndex) GetValByTID(tid uint32) []byte {`
`182`	`182`	`}`
`183`	`183`
`184`	`184`	`func (si *activeTokenIndex) GetTIDsByTokenExpr(t parser.Token) ([]uint32, error) {`
`185`		`- return si.tokenList.FindPattern(si.ctx, t, nil)`
	`185`	`+ return si.tokenList.FindPattern(si.ctx, t)`
`186`	`186`	`}`
`187`	`187`
`188`	`188`	`func (si *activeTokenIndex) GetLIDsFromTIDs(tids []uint32, _ lids.Counter, minLID, maxLID uint32, order seq.DocsOrder) []node.Node {`
Original file line number	Diff line number	Diff line change
`@@ -184,7 +184,7 @@ func (tl TokenList) getTokenProvider(field string) activeTokenProvider {`
`184`	`184`	`}`
`185`	`185`	`}`
`186`	`186`
`187`		`-func (tl *TokenList) FindPattern(ctx context.Context, t parser.Token, tids []uint32) ([]uint32, error) {`
	`187`	`+func (tl *TokenList) FindPattern(ctx context.Context, t parser.Token) ([]uint32, error) {`
`188`	`188`	`field := parser.GetField(t)`
`189`	`189`	`tp := tl.getTokenProvider(field)`
`190`	`190`	`tids, err := pattern.Search(ctx, t, tp)`
Original file line number	Diff line number	Diff line change
`@@ -218,7 +218,6 @@ func TestGrpcV1_FetchLive(t *testing.T) {`
`218`	`218`	`}`
`219`	`219`
`220`	`220`	`for _, tt := range tests {`
`221`		`- tt := tt`
`222`	`221`	`t.Run(tt.name, func(t *testing.T) {`
`223`	`222`	`t.Parallel()`
`224`	`223`	`r := require.New(t)`
Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,6 @@ func TestMain(m *testing.M) {`
`103`	`103`	`}`
`104`	`104`
`105`	`105`	`func runGRPCServerWithClient(apiServer seqproxyapi.SeqProxyApiServer) (seqproxyapi.SeqProxyApiClient, func()) {`
`106`		`- ctx := context.Background()`
`107`	`106`	`lis := bufconn.Listen(10 * 1024 * 1024)`
`108`	`107`	`server := grpc.NewServer(`
`109`	`108`	`grpc.StatsHandler(&tracing.ServerHandler{}),`
`@@ -121,9 +120,8 @@ func runGRPCServerWithClient(apiServer seqproxyapi.SeqProxyApiServer) (seqproxya`
`121`	`120`	`}`
`122`	`121`	`server.Stop()`
`123`	`122`	`}`
`124`		`- conn, err := grpc.DialContext(`
`125`		`- ctx,`
`126`		`- "",`
	`123`	`+ conn, err := grpc.NewClient(`
	`124`	`+ "localhost",`
`127`	`125`	`grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {`
`128`	`126`	`return lis.Dial()`
`129`	`127`	`}),`
Original file line number	Diff line number	Diff line change
`@@ -256,7 +256,7 @@ func (r *esBulkDocReader) readDoc() ([]byte, bool, error) {`
`256`	`256`
`257`	`257`	`// If this is a line that exceeds maxDocumentSize, skip it.`
`258`	`258`	`for isPrefix {`
`259`		`- doc, isPrefix, err = r.r.ReadLine()`
	`259`	`+ _, isPrefix, err = r.r.ReadLine()`
`260`	`260`	`if err != nil {`
`261`	`261`	`return nil, true, err`
`262`	`262`	`}`
Original file line number	Diff line number	Diff line change
`@@ -47,38 +47,37 @@ type Ingestor struct {`
`47`	`47`	`isStopped atomic.Bool`
`48`	`48`	`}`
`49`	`49`
`50`		`-func clientsFromConfig(ctx context.Context, config search.Config) (map[string]storeapi.StoreApiClient, error) {`
	`50`	`+func clientsFromConfig(config search.Config) (map[string]storeapi.StoreApiClient, error) {`
`51`	`51`	`clients := map[string]storeapi.StoreApiClient{}`
`52`		`- if err := appendClients(ctx, clients, config.HotStores.Shards); err != nil {`
	`52`	`+ if err := appendClients(clients, config.HotStores.Shards); err != nil {`
`53`	`53`	`return nil, err`
`54`	`54`	`}`
`55`	`55`	`if config.HotReadStores != nil {`
`56`		`- if err := appendClients(ctx, clients, config.HotReadStores.Shards); err != nil {`
	`56`	`+ if err := appendClients(clients, config.HotReadStores.Shards); err != nil {`
`57`	`57`	`return nil, err`
`58`	`58`	`}`
`59`	`59`	`}`
`60`	`60`	`if config.WriteStores != nil {`
`61`		`- if err := appendClients(ctx, clients, config.WriteStores.Shards); err != nil {`
	`61`	`+ if err := appendClients(clients, config.WriteStores.Shards); err != nil {`
`62`	`62`	`return nil, err`
`63`	`63`	`}`
`64`	`64`	`}`
`65`	`65`	`if config.ReadStores != nil {`
`66`		`- if err := appendClients(ctx, clients, config.ReadStores.Shards); err != nil {`
	`66`	`+ if err := appendClients(clients, config.ReadStores.Shards); err != nil {`
`67`	`67`	`return nil, err`
`68`	`68`	`}`
`69`	`69`	`}`
`70`	`70`	`return clients, nil`
`71`	`71`	`}`
`72`	`72`
`73`		`-func appendClients(ctx context.Context, clients map[string]storeapi.StoreApiClient, shards [][]string) error {`
	`73`	`+func appendClients(clients map[string]storeapi.StoreApiClient, shards [][]string) error {`
`74`	`74`	`for _, shard := range shards {`
`75`	`75`	`for _, replica := range shard {`
`76`	`76`	`if _, has := clients[replica]; has {`
`77`	`77`	`continue`
`78`	`78`	`}`
`79`	`79`	`// this doesn't block, and if store is down, it will try to reconnect in background`
`80`		`- conn, err := grpc.DialContext(`
`81`		`- ctx,`
	`80`	`+ conn, err := grpc.NewClient(`
`82`	`81`	`replica,`
`83`	`82`	`grpc.WithTransportCredentials(insecure.NewCredentials()),`
`84`	`83`	`grpc.WithStatsHandler(&tracing.ClientHandler{}),`
`@@ -132,7 +131,7 @@ func NewIngestor(config IngestorConfig, store storeapiclient.Store) (Ingestor,`
`132`	`131`	`config.Bulk.HotStores = stores.NewStoresFromString("memory", 1)`
`133`	`132`	`config.Search.HotStores = stores.NewStoresFromString("memory", 1)`
`134`	`133`	`} else {`
`135`		`- clients, err = clientsFromConfig(ctx, config.Search)`
	`134`	`+ clients, err = clientsFromConfig(config.Search)`
`136`	`135`	`if err != nil {`
`137`	`136`	`cancel()`
`138`	`137`	`return nil, fmt.Errorf("initialize clients: %s", err)`