From f7302ea506d4b25653b975b1d0cf32ca6466c00c Mon Sep 17 00:00:00 2001 From: "T. von Dein" Date: Fri, 29 May 2026 10:19:18 +0200 Subject: [PATCH] more search enhancements (#22) --- cmd/root.go | 29 ++++++ cmd/search.go | 53 +++++++++-- pkg/cfg/config.go | 56 ++++++++++-- pkg/es/search.go | 160 ++++++++++++++++++++++++++++---- pkg/es/search_filter.go | 196 ++++++++++++++++++++++++++++++---------- pkg/printer/doc.go | 45 +++++++++ 6 files changed, 456 insertions(+), 83 deletions(-) create mode 100644 pkg/printer/doc.go diff --git a/cmd/root.go b/cmd/root.go index 6916dec..f7d7bea 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -22,6 +22,7 @@ import ( "os" "codeberg.org/scip/esctl/pkg/cfg" + "codeberg.org/scip/esctl/pkg/es" "codeberg.org/scip/esctl/pkg/log" "github.com/urfave/cli/v3" @@ -54,6 +55,12 @@ func Main() int { Sources: cli.EnvVars("ES_DEBUG"), Destination: &conf.Debug, }, + &cli.BoolFlag{ + Name: "debug-http", + Value: false, + Usage: "enable HTTP debugging", + Destination: &conf.DebugHTTP, + }, &cli.StringFlag{ Name: "config", Aliases: []string{"c"}, @@ -89,6 +96,7 @@ func Main() int { Doc(conf), Repl(conf), Version(conf), + Debug(conf), }, Before: func(ctx context.Context, cmd *cli.Command) (context.Context, error) { @@ -122,3 +130,24 @@ func Version(conf *cfg.Config) *cli.Command { }, } } + +func Debug(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "debug", + Usage: "developer only", + + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "index", + Usage: "index to search within", + Sources: cli.EnvVars("ES_INDEX"), + Destination: &conf.Index, + Aliases: []string{"i"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + return es.Debug(conf) + }, + } +} diff --git a/cmd/search.go b/cmd/search.go index 5bed9f1..7a95897 100644 --- a/cmd/search.go +++ b/cmd/search.go @@ -36,12 +36,23 @@ func Search(conf *cfg.Config) *cli.Command { might be one of: =: Must match !=: Must not match - ?: Should match You can omit a field spec and thereby search across all fields. +By default all queries contribute to matches (logical AND), use +-O to apply a logical OR operator. + You can also search multiple fields by separating them with comma, eg: -user,group=root`, +user,group=root + +Use filters to further restrict results, they must match literally. + +For datetime range format refer to: +https://www.elastic.co/docs/reference/elasticsearch/rest-apis/common-options#date-math + +For timestamp formats refer to: +https://www.elastic.co/docs/reference/elasticsearch/mapping-reference/mapping-date-format +`, Flags: []cli.Flag{ &cli.StringFlag{ @@ -53,17 +64,17 @@ user,group=root`, }, &cli.IntFlag{ Name: "from", - Usage: "show results FROM (default 0)", + Usage: "show results starting at ", Destination: &conf.From, Value: 0, Aliases: []string{"f"}, }, &cli.IntFlag{ - Name: "to", - Usage: "show results to (default 10)", + Name: "len", + Usage: "number of results to show (-1: all[max:10k], caution: might be slow)", Destination: &conf.To, - Value: 10, - Aliases: []string{"t"}, + Value: 20, + Aliases: []string{"l"}, }, &cli.StringSliceFlag{ Name: "filter", @@ -77,12 +88,36 @@ user,group=root`, Destination: &conf.Path, Aliases: []string{"p"}, }, + &cli.StringFlag{ + Name: "timerange", + Usage: "field: to (e.g. @timestamp:2026-05-05 to 2026-05-15)", + Destination: &conf.Range, + Aliases: []string{"r"}, + }, + &cli.StringFlag{ + Name: "timestamp-format", + Usage: "a valid ES builtin timestamp or custom format", + Destination: &conf.TimestampFormat, + Value: "strict_date_hour_minute", + }, &cli.BoolFlag{ Name: "help-jsonpath", Usage: "show jsonPath help", Destination: &conf.Subhelp, Aliases: []string{"H"}, }, + &cli.BoolFlag{ + Name: "tail", + Usage: "follow search live, like tail -f", + Destination: &conf.Tail, + Aliases: []string{"T"}, + }, + &cli.BoolFlag{ + Name: "or", + Usage: "logical operator (default: and)", + Destination: &conf.Or, + Aliases: []string{"O"}, + }, }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -92,6 +127,10 @@ user,group=root`, args := cmd.Args() + if conf.To == -1 { + conf.To = 10000 + } + return es.Search(conf, args.Slice()) }, } diff --git a/pkg/cfg/config.go b/pkg/cfg/config.go index 8a2eed3..c10ca6d 100644 --- a/pkg/cfg/config.go +++ b/pkg/cfg/config.go @@ -17,10 +17,13 @@ along with this program. If not, see . package cfg import ( + "bytes" "context" "crypto/tls" + "encoding/json" "errors" "fmt" + "log/slog" "net/http" "os" @@ -60,11 +63,16 @@ type Config struct { Filter []string // search: -F Path string // search+doc sh: -p Subhelp bool // search+doc sh: -H + Tail bool // search: -f [tail] + Or bool // search: -O + Range string // search: -r + TimestampFormat string // search: --timestamp-format Exclude string // cluster compare: -e (regexp) All, Verbose bool // cluster status: -a -v Persistent, Transient, Default bool // -p -t -D cluster settings set Force bool // ccr follower renew: -f HaveJQ bool // determined at runtime by ourselfes + DebugHTTP bool // root: --debug-http } func NewConfig() *Config { @@ -200,18 +208,26 @@ func (conf *Config) LoadConfig() error { return nil } +func (conf *Config) getTransport() elastictransport.Option { + transport := &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + } + + if conf.DebugHTTP { + return elastictransport.WithTransport( + &DebugTransport{Transport: transport}, + ) + } + + return elastictransport.WithTransport(transport) +} + func (conf *Config) SetupES() error { for _, cluster := range conf.Clusters { es, err := elasticsearch.NewTyped( elasticsearch.WithAddresses(cluster.Uri), elasticsearch.WithBasicAuth(cluster.User, cluster.Pass), - elasticsearch.WithTransportOptions( - elastictransport.WithTransport( - &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - }, - ), - ), + elasticsearch.WithTransportOptions(conf.getTransport()), ) if err != nil { @@ -224,6 +240,32 @@ func (conf *Config) SetupES() error { return nil } +// used to print uri, path and body of a request made by the go-client +type DebugTransport struct { + Transport http.RoundTripper +} + +func (t *DebugTransport) RoundTrip(req *http.Request) (*http.Response, error) { + buf := new(bytes.Buffer) + + body, _ := req.GetBody() + + _, err := buf.ReadFrom(body) + if err != nil { + return nil, err + } + + var pretty bytes.Buffer + err = json.Indent(&pretty, buf.Bytes(), "", "\t") + if err != nil { + return nil, fmt.Errorf("json parse error: %s", err) + } + + slog.Info("req", "host", req.URL.Host, "uri", req.URL.Path, "body", pretty.String(), "bodyline", buf.String()) + + return t.Transport.RoundTrip(req) +} + func fileExists(filename string) bool { info, err := os.Stat(filename) diff --git a/pkg/es/search.go b/pkg/es/search.go index 61078a1..8d3cec3 100644 --- a/pkg/es/search.go +++ b/pkg/es/search.go @@ -19,10 +19,22 @@ package es import ( "context" "fmt" + "log" "log/slog" + "strings" + "time" "codeberg.org/scip/esctl/pkg/cfg" - "github.com/tidwall/gjson" + "codeberg.org/scip/esctl/pkg/printer" + "github.com/alecthomas/repr" + "github.com/elastic/go-elasticsearch/v9/typedapi/core/search" + "github.com/elastic/go-elasticsearch/v9/typedapi/esdsl" + "github.com/elastic/go-elasticsearch/v9/typedapi/types" + "github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/sortorder" +) + +const ( + MAXPAGE = 5000 ) /* @@ -35,36 +47,146 @@ func Search(conf *cfg.Config, queries []string) error { search := conf.DefaultCluster.ES.Search(). Index(conf.Index) - if len(queries) > 0 { - req, err := prepareQuery(conf, queries) - if err != nil { - return err - } - - search.Request(req) + req, err := prepareQuery(conf, queries) + if err != nil { + return err } - res, err := search.Do(context.Background()) + search.Request(req) + + switch conf.Tail { + case true: + return searchTail(conf, search) + case false: + if conf.To > MAXPAGE { + return searchPit(conf, req) + } else { + return searchOnce(conf, search) + } + } + + return nil +} + +func Debug(conf *cfg.Config) error { + res, err := conf.DefaultCluster.ES.Search(). + Index(conf.Index). + Size(0). + Aggregations(map[string]types.Aggregations{ + "min_ts": *esdsl.NewMinAggregation().Field("@timestamp").AggregationsCaster(), + "max_ts": *esdsl.NewMaxAggregation().Field("@timestamp").AggregationsCaster(), + }). + Do(context.Background()) + if err != nil { + return err + } + + repr.Println(res) + + return nil +} + +func searchOnce(conf *cfg.Config, search *search.Search) error { + res, err := search. + From(conf.From). + Size(conf.To). + Do(context.Background()) + if err != nil { + if strings.Contains(err.Error(), "reason: all shards failed") { + return nil + } + return fmt.Errorf("failed to run search (esdsl): %s", err) } slog.Debug("ES result", "search", res) for _, hit := range res.Hits.Hits { - docjson := fmt.Sprintf(`{"id":%s, "score":%0.4f, "index":"%s", "source":%s}`, - *hit.Id_, - *hit.Score_, - hit.Index_, - hit.Source_) + printer.PrintDoc(conf, hit) + } - if conf.Path != "" { - value := gjson.Get(docjson, conf.Path) - fmt.Println(value.String()) - } else { - fmt.Println(docjson) + return nil +} + +// https://www.elastic.co/docs/reference/elasticsearch/clients/go/using-the-api/searching#_pit_search_after +func searchPit(conf *cfg.Config, req *search.Request) error { + ctx := context.Background() + pit, err := conf.DefaultCluster.ES.OpenPointInTime(conf.Index).KeepAlive("1m").Do(ctx) + if err != nil { + return fmt.Errorf("failed to open point-in-time request for search: %s", err) + } + defer func() { + _, err := conf.DefaultCluster.ES.ClosePointInTime().Id(pit.Id).Do(ctx) + if err != nil { + log.Fatalf("failed to close PIT: %s", err) + } + }() + + search := conf.DefaultCluster.ES.Search(). + Request(req). + Pit(esdsl.NewPointInTimeReference(). + Id(pit.Id). + KeepAlive(esdsl.NewDuration().String("1m"))). + Sort(esdsl.NewSortOptions(). + AddSortOption("_shard_doc", esdsl.NewFieldSort(sortorder.Asc))). + Size(conf.To) + + for { + res, err := search.Do(ctx) + if err != nil { + return fmt.Errorf("failed to run search (esdsl pit): %s", err) + } + + if len(res.Hits.Hits) == 0 { + break + } + + for _, hit := range res.Hits.Hits { + printer.PrintDoc(conf, hit) + } + + last := res.Hits.Hits[len(res.Hits.Hits)-1] + search = search.SearchAfterValues(last.Sort) + + if res.PitId != nil { + search = search.Pit(esdsl.NewPointInTimeReference(). + Id(*res.PitId). + KeepAlive(esdsl.NewDuration().String("1m"))) } } return nil } + +func searchTail(conf *cfg.Config, search *search.Search) error { + docs := map[string]int{} + + fmt.Println("enter ctrl-c to abort...") + + for { + res, err := search.Do(context.Background()) + if err != nil { + if strings.Contains(err.Error(), "reason: all shards failed") { + return nil + } + + return fmt.Errorf("failed to run search (esdsl): %s", err) + } + + slog.Debug("ES result", "search", res) + + for _, hit := range res.Hits.Hits { + _, exists := docs[*hit.Id_] + if exists { + continue + } + + printer.PrintDoc(conf, hit) + + docs[*hit.Id_] = 1 + } + + time.Sleep(100 * time.Millisecond) + } +} diff --git a/pkg/es/search_filter.go b/pkg/es/search_filter.go index d1715d1..986739b 100644 --- a/pkg/es/search_filter.go +++ b/pkg/es/search_filter.go @@ -17,13 +17,16 @@ along with this program. If not, see . package es import ( + "errors" "fmt" + "log/slog" "strings" "codeberg.org/scip/esctl/pkg/cfg" "github.com/elastic/go-elasticsearch/v9/typedapi/core/search" "github.com/elastic/go-elasticsearch/v9/typedapi/esdsl" "github.com/elastic/go-elasticsearch/v9/typedapi/types" + "github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/operator" ) const ( @@ -40,7 +43,22 @@ type filter struct { multi bool // message,title=foo => filter:foo, multi: []string{"message","title"} } -// build a new filter object +// Build a new filter object. We use this to build our elastic query +// out of it. We support differnt types of queries: +// +// - nop filter: no query at all, just return the first N documents. +// +// - simple ones like: "authenticated", which match across all fields +// +// - simple ones combined like "authenticated monitoring", if -O was set, +// apply an OR logic +// +// - specific field queries: "message=authenticated" (can be combined too like above) +// +// - and specific field queries with negation: "message!=authenticated" +// +// All queries support additional filters using -F field=value, which +// must match literally, and range filters using -r "@timestamp:2026-05-28T10:00:00 to now" func NewFilter(query string) (*filter, error) { var separator string var criteria int // we use the constants on top for this @@ -49,9 +67,6 @@ func NewFilter(query string) (*filter, error) { case strings.Contains(query, "!="): criteria = Fmustnot separator = "!=" - case strings.Contains(query, "?"): - criteria = Fshould - separator = "?" default: criteria = Fmust separator = "=" @@ -59,48 +74,30 @@ func NewFilter(query string) (*filter, error) { part := strings.Split(query, separator) if len(part) != 2 { - return nil, fmt.Errorf("search queries must be in the form fieldpattern where must be one of: =, !=, ?") + return nil, fmt.Errorf("search queries must be in the form fieldpattern where must be one of: = or !=") } - f := &filter{term: part[0], filter: part[1], criteria: criteria} + flt := &filter{term: part[0], filter: part[1], criteria: criteria} if strings.Contains(part[0], ",") { // a MultiMatchQuery, match across multiple fields at once multi := strings.Split(part[0], ",") - f.multi = true - f.mterm = multi + flt.multi = true + flt.mterm = multi } - return f, nil + return flt, nil } -// prepare q user search query and turn it into a proper search request -func prepareQuery(conf *cfg.Config, queries []string) (*search.Request, error) { - if len(queries) == 0 { - // nothing given, just return all docs, if any - return &search.Request{ - Query: esdsl.NewMatchAllQuery().QueryCaster(), - From: &conf.From, - Size: &conf.To, - }, nil - } - - if len(queries) == 1 && !strings.ContainsAny(queries[0], "!=?") { - // a general query w/o fields, search across all fields - return &search.Request{ - Query: esdsl.NewSimpleQueryStringQuery(queries[0]).QueryCaster(), - From: &conf.From, - Size: &conf.To, - }, nil - } - - // complex query, form a proper query struct - query := esdsl.NewBoolQuery() +// Build a complex query set for queries containing field[!]=pattern +func mkMatchQueries(queries []string, op operator.Operator) ([]types.QueryVariant, []types.QueryVariant, error) { + matchqueries := []types.QueryVariant{} + matchNotqueries := []types.QueryVariant{} for _, q := range queries { filter, err := NewFilter(q) if err != nil { - return nil, err + return nil, nil, err } // by default we match on a single field @@ -108,23 +105,92 @@ func prepareQuery(conf *cfg.Config, queries []string) (*search.Request, error) { if filter.multi { // ok, match across multiple given fields - match = esdsl.NewMultiMatchQuery(filter.filter).Fields(filter.mterm...) + match = esdsl.NewMultiMatchQuery(filter.filter).Fields(filter.mterm...).Operator(op) } - // apply logic - switch filter.criteria { - case Fmustnot: - query.MustNot(match) - case Fmust: - query.Must(match) - case Fshould: - query.Should(match) + if filter.criteria == Fmustnot { + matchNotqueries = append(matchNotqueries, match) + } else { + matchqueries = append(matchqueries, match) } } - // there might be boolean filters as well - filters := make([]types.QueryVariant, len(conf.Filter)) + return matchqueries, matchNotqueries, nil +} +// Prepare user search query and turn it into a proper search request +func prepareQuery(conf *cfg.Config, queries []string) (*search.Request, error) { + // logical operator for simple and multimatch queries + op := operator.Operator{Name: "AND"} + if conf.Or { + op = operator.Operator{Name: "OR"} + } + + query := esdsl.NewBoolQuery() + wholeQuery := strings.Join(queries, " ") + + switch { + case len(queries) == 0: + // nothing provided via ARGs, so match any docs + query.Must(esdsl.NewMatchAllQuery()) + + case !strings.ContainsAny(wholeQuery, "!="): + // simple query w/o any field[!]=pattern style + query.Must(esdsl.NewSimpleQueryStringQuery(wholeQuery).DefaultOperator(op)) + + default: + // a complex query + matchqueries, matchNotqueries, err := mkMatchQueries(queries, op) + if err != nil { + return nil, err + } + + // apply boolean logic + if conf.Or { + query.Should(matchqueries...) + } else { + // by default we use AND + query.Must(matchqueries...) + } + + if matchNotqueries != nil { + query.MustNot(matchNotqueries...) + } + } + + slog.Debug("complex query", "query", query) + + // there might be boolean or range filters like -Ffield=value as well + filters, err := addFilters(conf) + if err != nil { + return nil, err + } + + if filters != nil { + query.Filter(filters...) + } + + // this being fed into ES.Serach().Req() + return &search.Request{ + Query: query.QueryCaster(), + }, nil +} + +// Build a slice of filters, to be fed into query.Filter() including static ones +// like -Ffield=value and ranges like -r "@timestamp:2026-05-28T10:00:00 to now" +func addFilters(conf *cfg.Config) ([]types.QueryVariant, error) { + count := len(conf.Filter) + if conf.Range != "" { + count++ + } + + if count == 0 { + return nil, nil + } + + filters := make([]types.QueryVariant, count) + + // static filters for idx, filter := range conf.Filter { parts := strings.Split(filter, "=") if len(parts) != 2 { @@ -134,13 +200,43 @@ func prepareQuery(conf *cfg.Config, queries []string) (*search.Request, error) { filters[idx] = esdsl.NewTermQuery(parts[0], esdsl.NewFieldValue().String(parts[1])) } - if len(filters) > 0 { - query.Filter(filters...) + // range filters. format: @timestamp:2026-05-05 to 2026-05-15 + // see: https://www.elastic.co/docs/reference/elasticsearch/rest-apis/common-options#date-math + if conf.Range != "" { + parts := strings.SplitN(conf.Range, ":", 2) + if len(parts) != 2 { + return nil, errors.New("invalid range format, expected field:range") + } + + field := parts[0] + + parts = strings.Split(parts[1], " to ") + if len(parts) != 2 { + return nil, errors.New("invalid date range format, expected ' to '") + } + + from := parts[0] + to := parts[1] + + slog.Debug("time range filter", + "from", from, + "to", to, + "format", conf.TimestampFormat, + "count", count, + ) + + rng := esdsl.NewDateRangeQuery(field). + Gte(from). + Lte(to) + + if strings.Contains(parts[1], ":") { + // specific format not needed as long as there are no times specified + rng.Format(conf.TimestampFormat) + } + + filters[count-1] = rng + } - return &search.Request{ - Query: query.QueryCaster(), - From: &conf.From, - Size: &conf.To, - }, nil + return filters, nil } diff --git a/pkg/printer/doc.go b/pkg/printer/doc.go new file mode 100644 index 0000000..00f8055 --- /dev/null +++ b/pkg/printer/doc.go @@ -0,0 +1,45 @@ +/* +Copyright © 2026 Thomas von Dein + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ +package printer + +import ( + "fmt" + + "codeberg.org/scip/esctl/pkg/cfg" + "github.com/elastic/go-elasticsearch/v9/typedapi/types" + "github.com/tidwall/gjson" +) + +func PrintDoc(conf *cfg.Config, hit types.Hit) { + var score types.Float64 + if hit.Score_ != nil { + score = *hit.Score_ + } + + docjson := fmt.Sprintf(`{"id":"%s", "score":%0.4f, "index":"%s", "source":%s}`, + *hit.Id_, + score, + hit.Index_, + hit.Source_) + + if conf.Path != "" { + value := gjson.Get(docjson, conf.Path) + fmt.Println(value.String()) + } else { + fmt.Println(docjson) + } +}